1. Configuração do Ambiente

1.1. Ambiente Windows

1. Copie os diretórios pyspark e py4j para a pasta de instalação do Python.
2. Adicione os JARs adicionais necessários à pasta jars do Spark.
3. Configure o interpretador Python no PyCharm e ajuste as configurações de proxy da empresa, se aplicável.

1.2. Ambiente Linux

1. Copie os diretórios pyspark e py4j para a pasta de instalação do Python.
2. Adicione os JARs adicionais necessários à pasta jars do Spark. Exemplos:
   • mongo-java-driver-3.12.5.jar
   • mongo-spark-connector\_2.11-2.4.2.jar
3. Empacote as dependências em um arquivo ZIP dentro do seu projeto para evitar erros.
4. Copie o arquivo .py do PySpark a ser executado separadamente.
5. Ajuste a configuração do spark-env.sh para especificar a versão do Python: ``` export PYSPARK_PYTHON=/usr/bin/python3

6. Execute usando spark-submit: ``` /boot/clouddragen/spark/bin/spark-submit
   --master spark://xxx:7077,xxx:7077
   --py-files /boot/clouddragen/temp/crawler-spark.zip
   --executor-memory 10G
   /boot/clouddragen/temp/ReadMongodb.py

Exemplo Básico de PySpark

from pyspark import SparkConf, SparkContext

conf = SparkConf() \
   .setMaster("local[*]") \
   .setAppName("ContagemPalavras")
sc = SparkContext.getOrCreate(conf)

dados = [1, 2, 3, 4, 5]
rdd = sc.parallelize(dados)
print(rdd.collect())

2. Princípios de Funcionamento do PySpark

O Spark é primariamente desenvolvido em Scala e, para facilitar a integração com outros sistemas, partes de sua implementação utilizam Java (como o External Shuffle Service). Assim, o Spark opera dentro da Java Virtual Machine (JVM). Embora ofereça interfaces para Scala e Java, ele também disponibiliza APIs para Python e R, mantendo a independência do núcleo do Spark através de camadas de empacotamento. Este documento foca no funcionamento do PySpark. ### Arquitetura de Runtime do Spark

A arquitetura básica do Spark envolve um processo *Driver* e múltiplos processos *Executor*. O *Driver* gerencia o planejamento geral e a interface do usuário, enquanto os *Executors* executam as tarefas. O Spark pode ser implantado em diversos sistemas de gerenciamento de recursos como YARN e Mesos, ou em seu próprio modo Standalone. Um aplicativo Spark roda no *Driver*. O Spark encapsula o código do usuário em *Tasks*, que são enviadas aos *Executors* contendo a lógica de execução e os dados. Os *Executors* não executam o código do usuário diretamente. ### Arquitetura de Runtime do PySpark

Para preservar a arquitetura existente do Spark, o PySpark utiliza uma camada de API Python e a biblioteca Py4j para interagir com a JVM. Abaixo, um esquema que ilustra a interação:

Os componentes em branco na figura representam processos Python adicionais.

• No lado do Driver: Através do Py4j, as chamadas de métodos Python são mapeadas para métodos Java na JVM. Por exemplo, a instanciação de um SparkContext em Python resulta na instanciação de um SparkContext Scala/Java na JVM.
• No lado do Executor: A comunicação com a JVM é feita de forma diferente. As tarefas enviadas pelo Driver contêm a lógica de execução, que pode incluir funções ou lambdas Python serializadas. Como o Py4j não permite chamadas Java para Python, um processo Python separado é iniciado para cada tarefa. A comunicação entre a JVM e o processo Python do Executor ocorre via sockets para executar funções e lambdas Python.

O fluxo geral de interação entre as linguagens é o seguinte: ### Funcionamento no Driver

Ao submeter um script PySpark com spark-submit, os scripts Python e suas dependências são carregados, e recursos para o Driver são alocados. Um processo JVM é iniciado através do PythonRunner, que possui um método main. O método main do PythonRunner realiza duas ações principais: - Inicia o servidor Gateway do Py4j (Py4j GatewayServer).

• Executa o script Python do usuário através de um processo Java. O script Python, ao iniciar, instancia o SparkContext Python. Durante essa instanciação:
  • Um cliente Gateway do Py4j (Py4j GatewayClient) é instanciado para conectar ao GatewayServer na JVM. Todas as chamadas subsequentes de Python para Java usarão este Gateway.
  • Utilizando o Gateway do Py4j, o objeto SparkContext é instanciado na JVM.

Após a inicialização do SparkContext, o Driver aloca recursos para os *Executors* e começa o agendamento das tarefas. Quando um método de ação (action) é encontrado no script Python, um Job é submetido. Essa submissão é feita chamando o método runJob da classe PythonRDD na JVM via Py4j. Após a conclusão do Job, a JVM abre um socket local para que o processo Python possa buscar os resultados. A interação no Driver pode ser visualizada assim: Para grandes volumes de dados, como variáveis de broadcast, a comunicação entre os processos Python e JVM pode ocorrer através do sistema de arquivos local para otimizar a transferência de dados. ### Funcionamento no Executor

Considerando o cenário Spark on YARN, quando o Driver solicita recursos de Executor, um processo JVM é iniciado (CoarseGrainedExecutorBackend). Este processo inicia serviços necessários e aguarda o envio de tarefas do Driver. Não há processos Python ativos no Executor até que uma tarefa seja recebida. O processo de execução de uma tarefa no Executor segue estes passos: 1. Ao receber uma tarefa, o Executor chama launchTask, que eventualmente invoca o método compute do PythonRDD para processar uma partição de dados. 2. Se um processo de daemon Python (pyspark.daemon) não estiver em execução no Executor, um é iniciado. Cada Executor possui apenas um pyspark.daemon. Caso contrário, uma conexão via socket é estabelecida com o daemon existente. O pyspark.daemon solicita a criação de um processo pyspark.worker para executar a função ou lambda Python definida pelo usuário. O pyspark.daemon gerencia múltiplos pyspark.workers, onde cada worker processa uma tarefa simultaneamente. 3. Um thread separado alimenta o processo pyspark.worker com os dados, e o worker executa a função/lambda Python. 4. Enquanto os dados são alimentados, os resultados do pyspark.worker são recuperados via socket.

A interação no Executor pode ser visualizada assim: Em resumo, o PySpark utiliza o Py4j para permitir que o Python controle a execução de aplicações Spark baseadas em JVM. A comunicação de resultados da JVM para Python é realizada via sockets locais. Essa arquitetura, embora mantenha a independência do núcleo do Spark, pode gerar sobrecarga de performance devido à frequente comunicação entre os processos JVM e Python, especialmente em cenários de Big Data. Para aplicações de machine learning em larga escala ou streaming, é recomendável o uso de Scala/Java. PySpark é mais adequado para tarefas offline de menor escala onde a rapidez de desenvolvimento é prioritária. 3. Fontes de Dados no PySpark

3.1. Leitura do MongoDB

Para ler dados do MongoDB e criar um DataFrame: ``` from pyspark.sql import SparkSession

Importar outras dependências se necessário, como logging

input_uri = "mongodb://ip:27017/S3Libmaster.s3apis"

spark = SparkSession
.builder
.appName("LeituraMongo")
.config("spark.mongodb.input.uri", input_uri)
.config("spark.jars.packages", "org.mongodb.spark:mongo-spark-connector_2.11:2.4.2")
.getOrCreate()

Descomente para definir o nível de log

spark.sparkContext.setLogLevel('ERROR')

dataframe_mongo = spark.read.format('com.mongodb.spark.sql.DefaultSource').load()

dataframe_mongo.registerTempView("s3apis")

resultado_sql = spark.sql("SELECT fullName FROM s3apis WHERE fork = 'false' AND scanned=1")

spark.stop()

### 3.2. Leitura do MySQL

 Para ler dados do MySQL e criar um DataFrame: ```
from pyspark.sql import SparkSession
from pyspark.sql import SQLContext

spark = SparkSession.builder.appName("LeituraMySQL") \
    .master("local[10]") \
    .config('spark.some.config.option0', 'some-value') \
    .getOrCreate()
sqlContext = SQLContext(spark)

dataframe_mysql = sqlContext.read.format("jdbc") \
    .options(url="jdbc:mysql://IP:3306/ghtorrent",
             driver="com.mysql.jdbc.Driver",
             dbtable="(SELECT * FROM projects LIMIT 10000000) projects",
             user="seu_usuario",
             password="sua_senha") \
    .load()

dataframe_mysql.registerTempView("projects")

resultado_sql_mysql = spark.sql("SELECT id, url FROM projects WHERE language IS NOT NULL AND forked_from IS NULL AND deleted = 0")

# Descomente para coletar os resultados
# resultado_sql_mysql.collect()

print(resultado_sql_mysql.count())

spark.stop()