Construindo um pipeline de dados com PySpark

Quando os volumes de dados crescem além do que cabe na memória de uma única máquina, o Apache Spark se torna uma das ferramentas mais poderosas do arsenal de um engenheiro de dados. Neste post vou mostrar como estruturar um pipeline de ETL simples, porém realista, usando a API em Python — o PySpark.

O que vamos construir

Nosso pipeline segue a clássica arquitetura em três camadas (medallion):

Bronze — ingestão dos dados brutos, sem transformação.
Silver — dados limpos, tipados e deduplicados.
Gold — tabelas agregadas, prontas para consumo analítico.

1. Criando a SparkSession

Todo job PySpark começa com uma SparkSession, o ponto de entrada para trabalhar com DataFrames:

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .appName("pipeline-vendas")
    .config("spark.sql.shuffle.partitions", "200")
    .getOrCreate()
)

2. Camada Bronze — ingestão

Aqui apenas lemos os arquivos brutos (CSV, JSON, etc.) e os persistimos em Parquet, um formato colunar muito mais eficiente:

df_raw = (
    spark.read
    .option("header", True)
    .csv("s3://meu-bucket/raw/vendas/")
)

df_raw.write.mode("overwrite").parquet("s3://meu-bucket/bronze/vendas/")

3. Camada Silver — limpeza

Nesta etapa convertemos tipos, removemos duplicatas e filtramos registros inválidos:

from pyspark.sql import functions as F

df_silver = (
    spark.read.parquet("s3://meu-bucket/bronze/vendas/")
    .withColumn("valor", F.col("valor").cast("double"))
    .withColumn("data", F.to_date("data", "yyyy-MM-dd"))
    .dropDuplicates(["id_venda"])
    .filter(F.col("valor") > 0)
)

df_silver.write.mode("overwrite").parquet("s3://meu-bucket/silver/vendas/")

Dica: sempre que possível, particione a escrita por uma coluna de baixa cardinalidade (como data ou regiao). Isso acelera drasticamente as leituras filtradas posteriores.

4. Camada Gold — agregação

Por fim, calculamos as métricas de negócio que alimentarão dashboards e relatórios:

df_gold = (
    df_silver
    .groupBy("regiao", "data")
    .agg(
        F.sum("valor").alias("faturamento"),
        F.countDistinct("id_cliente").alias("clientes")
    )
)

df_gold.write.mode("overwrite").partitionBy("data").parquet(
    "s3://meu-bucket/gold/vendas_diarias/"
)

Boas práticas de performance

Técnica	Por quê
Usar Parquet/Delta	Leitura colunar + compressão reduzem I/O.
Particionar a escrita	Permite partition pruning nas consultas.
Evitar `collect()`	Traz todos os dados para o driver e pode estourar a memória.
`cache()` com cuidado	Útil para DataFrames reusados, mas consome memória dos executores.

Conclusão

Com pouco mais de 30 linhas construímos um pipeline em camadas que escala de gigabytes a terabytes apenas ajustando o cluster. A partir daqui você pode evoluir para Delta Lake (transações ACID), orquestrar com Airflow e adicionar testes de qualidade de dados.

Tem dúvidas ou sugestões? Fique à vontade para entrar em contato. 🚀