Como usar o Apache Spark para processamento de grandes volumes de dados

1. Fundamentos do Apache Spark e seu ecossistema

1.1. O que é o Apache Spark

Apache Spark é um motor de processamento unificado e de código aberto para análise de dados em larga escala. Desenvolvido originalmente na UC Berkeley em 2009, tornou-se um dos projetos mais ativos da Apache Software Foundation. Sua principal vantagem sobre o Hadoop MapReduce é a capacidade de realizar processamento em memória, alcançando velocidades até 100 vezes maiores para certas cargas de trabalho. Casos de uso típicos incluem ETL (Extract, Transform, Load), análise exploratória de dados, machine learning em escala e processamento de streams em tempo real.

1.2. Componentes principais

O ecossistema Spark é composto por cinco bibliotecas principais:
- Spark Core: base do framework, responsável pelo gerenciamento de memória, tolerância a falhas e agendamento de tarefas.
- Spark SQL: módulo para processamento de dados estruturados usando consultas SQL ou DataFrames.
- Spark Streaming: processamento de fluxos de dados em tempo real (legado, substituído pelo Structured Streaming).
- MLlib: biblioteca de machine learning escalável com algoritmos como regressão, classificação e clustering.
- GraphX: API para processamento de grafos e algoritmos de análise de redes.

1.3. Modos de implantação

O Spark pode ser executado em diversos ambientes:
- Local: ideal para desenvolvimento e testes com local[*].
- Standalone: cluster gerenciado pelo próprio Spark.
- YARN: integração com Hadoop YARN para gerenciamento de recursos.
- Kubernetes: suporte nativo para orquestração de contêineres.
- Databricks: plataforma gerenciada na nuvem que abstrai a complexidade do cluster.

2. Arquitetura de execução e conceitos essenciais

2.1. Driver, workers, executors e cluster manager

A arquitetura do Spark segue um modelo mestre-escravo:
- Driver: processo principal que executa a função main() e cria o SparkContext.
- Cluster Manager: gerencia os recursos do cluster (YARN, Mesos, Kubernetes).
- Workers: nós que executam processos workers.
- Executors: processos JVM que executam tarefas e armazenam dados em cache.

2.2. RDDs (Resilient Distributed Datasets)

RDDs são a abstração fundamental do Spark, representando uma coleção distribuída e imutável de objetos. Características principais:
- Transformações: operações lazy que criam novos RDDs (map, filter, flatMap).
- Ações: operações que disparam a execução (count, collect, saveAsTextFile).

# Exemplo de criação e operação com RDD
rdd = spark.sparkContext.parallelize([1, 2, 3, 4, 5])
rdd_filtered = rdd.filter(lambda x: x > 2)
resultado = rdd_filtered.count()  # Ação que dispara execução

2.3. DAG e lazy evaluation

O Spark constrói um grafo acíclico dirigido (DAG) de estágios para cada job. A avaliação lazy significa que as transformações são registradas em um grafo de linhagem, mas só executadas quando uma ação é chamada. Isso permite otimizações como:
- Pipelining: fusão de transformações estreitas.
- Shuffle minimization: reorganização de operações para reduzir transferência de dados.

3. Processamento de dados com Spark DataFrames e SQL

3.1. Criando DataFrames a partir de fontes diversas

DataFrames são a API moderna para processamento de dados estruturados, oferecendo maior desempenho que RDDs.

# Leitura de CSV
df_csv = spark.read.format("csv") \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .load("caminho/arquivo.csv")

# Leitura de Parquet (formato otimizado para Spark)
df_parquet = spark.read.parquet("caminho/dados.parquet")

# Leitura de JSON
df_json = spark.read.json("caminho/dados.json")

3.2. Transformações comuns

# Seleção e filtragem
df_clientes = spark.read.parquet("clientes.parquet")
df_sp = df_clientes.filter(df_clientes.estado == "SP") \
    .select("nome", "idade", "cidade")

# Agregação e ordenação
df_agregado = df_sp.groupBy("cidade") \
    .agg(avg("idade").alias("media_idade")) \
    .orderBy(col("media_idade").desc())

# Join entre DataFrames
df_vendas = spark.read.parquet("vendas.parquet")
df_resultado = df_vendas.join(df_sp, "cliente_id", "inner")

3.3. Spark SQL

Permite consultar DataFrames usando SQL padrão:

# Registrar DataFrame como tabela temporária
df_clientes.createOrReplaceTempView("clientes")

# Consulta SQL
resultado_sql = spark.sql("""
    SELECT cidade, AVG(idade) as media_idade
    FROM clientes
    WHERE estado = 'SP'
    GROUP BY cidade
    HAVING COUNT(*) > 100
    ORDER BY media_idade DESC
""")

4. Otimização de performance e tuning

4.1. Particionamento adequado

O número de partições influencia diretamente o paralelismo e o shuffle:

# Reparticionamento
df_reparticionado = df.repartition(200)

# Coalesce (reduz partições sem shuffle)
df_coalesced = df.coalesce(50)

# Evitar skew ajustando chaves de join
df_ajustado = df.withColumn("chave_hash", hash(col("chave")) % 100)

4.2. Cache e persistência

Armazenar DataFrames em memória acelera consultas repetitivas:

# Cache em memória (padrão: MEMORY_ONLY)
df_cached = df.cache()

# Persistência com diferentes níveis
from pyspark import StorageLevel
df_persisted = df.persist(StorageLevel.MEMORY_AND_DISK)

# Remover do cache
df_cached.unpersist()

4.3. Configurações críticas

# Configurações no SparkSession
spark = SparkSession.builder \
    .appName("Otimizacao") \
    .config("spark.sql.shuffle.partitions", "200") \
    .config("spark.executor.memory", "8g") \
    .config("spark.dynamicAllocation.enabled", "true") \
    .config("spark.dynamicAllocation.minExecutors", "2") \
    .config("spark.dynamicAllocation.maxExecutors", "20") \
    .getOrCreate()

5. Processamento em streaming com Spark Streaming e Structured Streaming

5.1. Diferenças entre DStreams e Structured Streaming

  • DStreams (Spark Streaming): API baseada em RDD, processa microlotes com latência de segundos.
  • Structured Streaming: API baseada em DataFrames, oferece processamento contínuo com latência sub-segundo e garantias exactly-once.

5.2. Configurando pipeline de streaming com Kafka

# Leitura de stream do Kafka
df_stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "topic_vendas") \
    .load()

# Processamento em tempo real
df_processado = df_stream.selectExpr(
    "CAST(value AS STRING) as json_string",
    "timestamp"
).select(
    from_json(col("json_string"), schema_vendas).alias("dados"),
    col("timestamp")
).select("dados.*", "timestamp")

# Escrita em console para debug
query = df_processado.writeStream \
    .outputMode("append") \
    .format("console") \
    .start()

5.3. Garantias de entrega

Para exactly-once, configure checkpointing e sinks idempotentes:

# Checkpointing para recuperação de falhas
query = df_processado.writeStream \
    .outputMode("append") \
    .format("parquet") \
    .option("path", "output/streaming") \
    .option("checkpointLocation", "checkpoints/streaming") \
    .start()

6. Integração com ecossistema de dados moderno

6.1. Delta Lake

Delta Lake adiciona transações ACID e time travel:

# Escrita em tabela Delta
df.write.format("delta").mode("overwrite").save("/delta/table")

# Leitura com time travel
df_historico = spark.read.format("delta") \
    .option("versionAsOf", 5) \
    .load("/delta/table")

# Leitura com timestamp
df_timestamp = spark.read.format("delta") \
    .option("timestampAsOf", "2024-01-01") \
    .load("/delta/table")

6.2. Conexão com data warehouses e data lakes

# Leitura de Snowflake
df_snowflake = spark.read.format("snowflake") \
    .option("sfUrl", "https://account.snowflakecomputing.com") \
    .option("sfUser", "user") \
    .option("sfPassword", "password") \
    .option("sfDatabase", "DB") \
    .option("sfSchema", "PUBLIC") \
    .option("dbtable", "TABELA") \
    .load()

# Escrita em S3 (com Hadoop AWS)
df.write.parquet("s3a://bucket/data/")

6.3. Orquestração com Apache Airflow

Jobs Spark podem ser submetidos via Airflow usando o SparkSubmitOperator:

# Exemplo de DAG no Airflow
from airflow.providers.apache.spark.operators.spark_submit import SparkSubmitOperator

submit_spark_job = SparkSubmitOperator(
    task_id='submit_spark_job',
    application='/path/to/script.py',
    conn_id='spark_default',
    conf={
        'spark.executor.memory': '8g',
        'spark.sql.shuffle.partitions': '200'
    },
    dag=dag
)

7. Boas práticas e padrões de projeto para produção

7.1. Estruturação de jobs

Modularize seu código em funções e classes reutilizáveis:

# Módulo de transformações
class Transformacoes:
    @staticmethod
    def limpar_dados(df):
        return df.dropna(subset=["coluna_importante"]) \
                 .filter(col("valor") > 0)

    @staticmethod
    def enriquecer(df):
        return df.withColumn("categoria", 
            when(col("valor") > 1000, "Alto")
            .otherwise("Baixo"))

# Uso no job principal
df_limpo = Transformacoes.limpar_dados(df_bruto)
df_enriquecido = Transformacoes.enriquecer(df_limpo)

7.2. Gerenciamento de dependências

Use pacotes Spark para bibliotecas externas:

# Submissão com pacotes
spark-submit \
  --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0 \
  --py-files dependencias.zip \
  job_principal.py

7.3. Estratégias para evitar erros comuns

# Evitar OutOfMemory: broadcast joins para tabelas pequenas
from pyspark.sql.functions import broadcast
df_resultado = df_grande.join(broadcast(df_pequena), "chave")

# Otimizar serialização: usar Kryo
spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

# Ajustar garbage collection
spark.conf.set("spark.executor.extraJavaOptions", 
    "-XX:+UseG1GC -XX:InitiatingHeapOccupancyPercent=35")

Referências