Otimização do Apache Spark

Padrões de produção para otimizar jobs do Apache Spark, incluindo estratégias de particionamento, gerenciamento de memória, otimização de shuffle e ajuste de performance.

Não use esta habilidade quando

• A tarefa não estiver relacionada à otimização do Apache Spark
• Você precisar de um domínio ou ferramenta diferente fora deste escopo

Instruções

• Clarifique objetivos, restrições e entradas necessárias.
• Aplique melhores práticas relevantes e valide os resultados.
• Forneça passos acionáveis e verificação.

Use esta habilidade quando

• Otimizando jobs Spark lentos
• Ajustando a configuração de memória e executores
• Implementando estratégias eficientes de particionamento
• Depurando problemas de performance do Spark
• Escalando pipelines Spark para grandes conjuntos de dados
• Reduzindo shuffle e assimetria de dados (data skew)

Conceitos Principais

1. Modelo de Execução do Spark

Driver Program
    ↓
Job (triggered by action)
    ↓
Stages (separated by shuffles)
    ↓
Tasks (one per partition)

2. Fatores Chave de Performance

Início Rápido

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

# Create optimized Spark session
spark = (SparkSession.builder
    .appName("OptimizedJob")
    .config("spark.sql.adaptive.enabled", "true")
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true")
    .config("spark.sql.adaptive.skewJoin.enabled", "true")
    .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    .config("spark.sql.shuffle.partitions", "200")
    .getOrCreate())

# Read with optimized settings
df = (spark.read
    .format("parquet")
    .option("mergeSchema", "false")
    .load("s3://bucket/data/"))

# Efficient transformations
result = (df
    .filter(F.col("date") >= "2024-01-01")
    .select("id", "amount", "category")
    .groupBy("category")
    .agg(F.sum("amount").alias("total")))

result.write.mode("overwrite").parquet("s3://bucket/output/")

Padrões

Padrão 1: Particionamento Ideal

# Calculate optimal partition count
def calculate_partitions(data_size_gb: float, partition_size_mb: int = 128) -> int:
    """
    Optimal partition size: 128MB - 256MB
    Too few: Under-utilization, memory pressure
    Too many: Task scheduling overhead
    """
    return max(int(data_size_gb * 1024 / partition_size_mb), 1)

# Repartition for even distribution
df_repartitioned = df.repartition(200, "partition_key")

# Coalesce to reduce partitions (no shuffle)
df_coalesced = df.coalesce(100)

# Partition pruning with predicate pushdown
df = (spark.read.parquet("s3://bucket/data/")
    .filter(F.col("date") == "2024-01-01"))  # Spark pushes this down

# Write with partitioning for future queries
(df.write
    .partitionBy("year", "month", "day")
    .mode("overwrite")
    .parquet("s3://bucket/partitioned_output/"))

Padrão 2: Otimização de Join

from pyspark.sql import functions as F
from pyspark.sql.types import *

# 1. Broadcast Join - Junção de tabelas pequenas
# Melhor quando: Um lado < 10MB (configurável)
small_df = spark.read.parquet("s3://bucket/small_table/")  # < 10MB
large_df = spark.read.parquet("s3://bucket/large_table/")  # TBs

# Dica explícita de broadcast
result = large_df.join(
    F.broadcast(small_df),
    on="key",
    how="left"
)

# 2. Sort-Merge Join - Padrão para tabelas grandes
# Requer shuffle, mas lida com qualquer tamanho
result = large_df1.join(large_df2, on="key", how="inner")

# 3. Bucket Join - Pré-ordenado, sem shuffle no momento do join
# Escrever tabelas particionadas (bucketed)
(df.write
    .bucketBy(200, "customer_id")
    .sortBy("customer_id")
    .mode("overwrite")
    .saveAsTable("bucketed_orders"))

# Juntar tabelas particionadas (sem shuffle!)
orders = spark.table("bucketed_orders")
customers = spark.table("bucketed_customers")  # Mesma contagem de buckets
result = orders.join(customers, on="customer_id")

# 4. Tratamento de Skew Join
# Habilitar otimização de skew join do AQE
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256MB")

# Salting manual para skew severo

…