M04 — Primeiro notebook Spark + Delta (Bronze → Silver)

Duração: 3h Pré-requisitos: M03 ok (Databricks lê do ADLS via SP). Objetivo: aprender o mínimo viável de PySpark + Delta Lake para transformar Bronze (parquet bruto) em Silver (Delta limpo, deduplicado, tipado, particionado).

Spark em uma página

Driver / Executors / Tasks

┌─────────────────────────────────────────────┐
│                   DRIVER                     │
│  - Roda seu código Python                    │
│  - Mantém SparkSession, planeja DAG          │
│  - Divide o job em stages e tasks            │
└──────────────────┬──────────────────────────┘
                   │ (envia tasks)
       ┌───────────┼───────────┐
       ▼           ▼           ▼
   EXECUTOR 1  EXECUTOR 2  EXECUTOR 3
   (JVM)       (JVM)       (JVM)
   - rodam tasks em paralelo
   - cada task processa 1 partition

Em Single Node (M03), driver + executor são o mesmo processo. Em cluster real, eles ficam em VMs diferentes.

Lazy evaluation

df = spark.read.parquet("...")        # NÃO leu nada ainda
df2 = df.filter("year > 2020")        # NÃO filtrou nada ainda
df3 = df2.groupBy("city").count()     # NÃO agregou nada ainda
df3.show()                             # AQUI dispara TUDO

Operações dividem-se em:

Transformations (lazy): select, filter, groupBy, join, withColumn…
Actions (eager): show, count, collect, write…

Wide vs Narrow

Narrow (sem shuffle): select, filter, withColumn. Cada partition gera 1 partition.
Wide (com shuffle, custoso): groupBy, join, distinct, repartition. Dados precisam atravessar a rede.

Otimização básica: filtre antes de agregar/joinar; reduza shuffles.

Partitions

DataFrame Spark = N partitions. Cada uma é processada por 1 task.
Default: spark.sql.shuffle.partitions = 200 (em cluster) / 8 (em local).
Regra de bolso: 128 MB por partition. Acima disso, Spark divide; muito pequeno, overhead supera ganho.

Delta Lake em uma página

Delta = Parquet + transaction log (_delta_log/). Te dá:

Recurso	Sem Delta	Com Delta
ACID	Não	Sim (commit atômico)
Upsert (`MERGE`)	Não	Sim
Time travel	Não	Sim (`VERSION AS OF 12`)
Schema evolution	Manual	Automático com flag
Vacuum (limpar files órfãos)	Manual	Comando nativo

Cada tabela é uma pasta:

silver/domain=sales/customer/
  ├─ _delta_log/
  │   ├─ 00000000000000000000.json    ← commit 0 (criação)
  │   ├─ 00000000000000000001.json    ← commit 1 (upsert)
  │   └─ 00000000000000000010.checkpoint.parquet
  ├─ part-00000-abc.snappy.parquet
  └─ part-00001-def.snappy.parquet

Parte 1 — Anatomia de um job Bronze → Silver

Vamos transformar o parquet de NYC Yellow Taxi (M02) em uma tabela Silver Delta com:

Tipos explícitos.
Coluna calculada trip_duration_min.
Filtro de outliers óbvios (corridas negativas, distâncias absurdas).
Deduplicação por pickup_datetime + pickup_location + dropoff_location.
Particionamento por pickup_date.

Schema esperado de Silver

Coluna	Tipo
vendor_id	INT
pickup_datetime	TIMESTAMP
dropoff_datetime	TIMESTAMP
pickup_date	DATE (partition key)
passenger_count	INT
trip_distance_km	DOUBLE
trip_duration_min	DOUBLE
pu_location_id	INT
do_location_id	INT
fare_amount	DOUBLE
tip_amount	DOUBLE
total_amount	DOUBLE
payment_type	STRING
_silver_ingested_at	TIMESTAMP

Parte 2 — Notebook completo

Crie no workspace 02_bronze_to_silver.py:

# Databricks notebook source
# MAGIC %md
# MAGIC # Bronze → Silver — NYC Yellow Taxi

# COMMAND ----------

# MAGIC %md
# MAGIC ## 1. Setup (credenciais + paths)

# COMMAND ----------

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

sp_appid  = dbutils.secrets.get("kv", "sp-appid")
sp_secret = dbutils.secrets.get("kv", "sp-secret")
sp_tenant = dbutils.secrets.get("kv", "sp-tenant")
storage   = dbutils.secrets.get("kv", "storage-account-name")

spark.conf.set(f"fs.azure.account.auth.type.{storage}.dfs.core.windows.net", "OAuth")
spark.conf.set(f"fs.azure.account.oauth.provider.type.{storage}.dfs.core.windows.net",
               "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider")
spark.conf.set(f"fs.azure.account.oauth2.client.id.{storage}.dfs.core.windows.net", sp_appid)
spark.conf.set(f"fs.azure.account.oauth2.client.secret.{storage}.dfs.core.windows.net", sp_secret)
spark.conf.set(f"fs.azure.account.oauth2.client.endpoint.{storage}.dfs.core.windows.net",
               f"https://login.microsoftonline.com/{sp_tenant}/oauth2/token")

bronze_path = f"abfss://bronze@{storage}.dfs.core.windows.net/source=nyctaxi/dataset=yellow/year=2024/month=01/"
silver_path = f"abfss://silver@{storage}.dfs.core.windows.net/domain=nyctaxi/yellow_trips/"

# COMMAND ----------

# MAGIC %md
# MAGIC ## 2. Read Bronze (raw)

# COMMAND ----------

df_bronze = spark.read.parquet(bronze_path)
print(f"Bronze: {df_bronze.count():,} linhas, {len(df_bronze.columns)} colunas")
df_bronze.printSchema()

# COMMAND ----------

# MAGIC %md
# MAGIC ## 3. Tipagem + colunas calculadas

# COMMAND ----------

df = (
    df_bronze
    .select(
        F.col("VendorID").cast(IntegerType()).alias("vendor_id"),
        F.col("tpep_pickup_datetime").cast(TimestampType()).alias("pickup_datetime"),
        F.col("tpep_dropoff_datetime").cast(TimestampType()).alias("dropoff_datetime"),
        F.col("passenger_count").cast(IntegerType()).alias("passenger_count"),
        (F.col("trip_distance") * 1.609344).cast(DoubleType()).alias("trip_distance_km"),  # mi → km
        F.col("PULocationID").cast(IntegerType()).alias("pu_location_id"),
        F.col("DOLocationID").cast(IntegerType()).alias("do_location_id"),
        F.col("fare_amount").cast(DoubleType()).alias("fare_amount"),
        F.col("tip_amount").cast(DoubleType()).alias("tip_amount"),
        F.col("total_amount").cast(DoubleType()).alias("total_amount"),
        F.col("payment_type").cast(StringType()).alias("payment_type"),
    )
    .withColumn("pickup_date", F.to_date("pickup_datetime"))
    .withColumn(
        "trip_duration_min",
        (F.col("dropoff_datetime").cast("long") - F.col("pickup_datetime").cast("long")) / 60.0
    )
    .withColumn("_silver_ingested_at", F.current_timestamp())
)

# COMMAND ----------

# MAGIC %md
# MAGIC ## 4. Limpeza (outliers + nulos críticos)

# COMMAND ----------

df_clean = (
    df
    .filter(F.col("pickup_datetime").isNotNull())
    .filter(F.col("dropoff_datetime") > F.col("pickup_datetime"))
    .filter(F.col("trip_duration_min").between(1, 24 * 60))       # 1 min ≤ dur ≤ 24h
    .filter(F.col("trip_distance_km").between(0.1, 200))          # corridas entre 100 m e 200 km
    .filter(F.col("fare_amount") >= 0)
    .filter(F.col("passenger_count").between(1, 6))
    .filter(F.year("pickup_date") == 2024)                        # garante mês esperado
)

descartados = df.count() - df_clean.count()
print(f"Descartados por outlier/null: {descartados:,}")

# COMMAND ----------

# MAGIC %md
# MAGIC ## 5. Deduplicação
# MAGIC
# MAGIC NYC Taxi traz registros duplicados ocasionais. Chave de unicidade:
# MAGIC pickup_datetime + pu_location_id + do_location_id + vendor_id.

# COMMAND ----------

from pyspark.sql.window import Window

w = Window.partitionBy(
    "pickup_datetime", "pu_location_id", "do_location_id", "vendor_id"
).orderBy(F.col("_silver_ingested_at").desc())

df_dedup = (
    df_clean
    .withColumn("_rn", F.row_number().over(w))
    .filter(F.col("_rn") == 1)
    .drop("_rn")
)

print(f"Após dedup: {df_dedup.count():,} linhas")

# COMMAND ----------

# MAGIC %md
# MAGIC ## 6. Escrita Delta (overwrite na 1ª vez; MERGE em produção)

# COMMAND ----------

(
    df_dedup
    .write
    .format("delta")
    .mode("overwrite")
    .option("overwriteSchema", "true")
    .partitionBy("pickup_date")
    .save(silver_path)
)

print(f"Escrito em: {silver_path}")

# COMMAND ----------

# MAGIC %md
# MAGIC ## 7. Validação

# COMMAND ----------

df_silver = spark.read.format("delta").load(silver_path)
print(f"Silver: {df_silver.count():,} linhas")
display(df_silver.limit(5))
display(
    df_silver
    .groupBy("pickup_date")
    .agg(F.count("*").alias("trips"), F.round(F.avg("fare_amount"), 2).alias("avg_fare"))
    .orderBy("pickup_date")
)

Importe e rode:

databricks workspace import 02_bronze_to_silver.py /Users/SEU_EMAIL/02_bronze_to_silver --language PYTHON --format SOURCE

Anexe ao cluster dev-single-node e Run all.

Parte 3 — Operações Delta que você precisa dominar

Time travel

# Versão N
df_v0 = spark.read.format("delta").option("versionAsOf", 0).load(silver_path)

# Timestamp
df_old = spark.read.format("delta").option("timestampAsOf", "2026-05-16T00:00:00").load(silver_path)

Histórico

%sql
DESCRIBE HISTORY delta.`abfss://silver@stdewdevbrs.dfs.core.windows.net/domain=nyctaxi/yellow_trips/`

MERGE (upsert) — padrão de Silver incremental

from delta.tables import DeltaTable

target = DeltaTable.forPath(spark, silver_path)

(
    target.alias("t")
    .merge(
        df_new_batch.alias("s"),
        "t.pickup_datetime = s.pickup_datetime AND t.pu_location_id = s.pu_location_id"
    )
    .whenMatchedUpdateAll()
    .whenNotMatchedInsertAll()
    .execute()
)

OPTIMIZE + ZORDER

Reorganiza arquivos pequenos em arquivos maiores; ZORDER agrupa dados similares.

%sql
OPTIMIZE delta.`<silver_path>`
  WHERE pickup_date >= '2024-01-01'
  ZORDER BY (pu_location_id);

VACUUM

Apaga arquivos órfãos (default: retém 7 dias para suportar time travel).

%sql
VACUUM delta.`<silver_path>` RETAIN 168 HOURS;   -- 168h = 7 dias (mínimo seguro)

Schema evolution

Adicionar coluna nova sem quebrar:

(df_new_with_extra_col
    .write
    .format("delta")
    .mode("append")
    .option("mergeSchema", "true")
    .save(silver_path))

Parte 4 — Spark UI + debug de performance

Durante o Run all, clique no link Spark Jobs / View que aparece embaixo da célula que disparou ação. Você vê:

Stages e suas tasks.
Tempo gasto em shuffle read/write, GC, computação.
DAG visualization.

Sinais de problema típicos:

Tasks de tamanho muito desigual → skew. Use salting ou broadcast join.
Muito tempo em GC → memória executor baixa.
Shuffle gigante em groupBy/join pequeno → faltou broadcast().

Parte 5 — Boas práticas que vão te poupar dor

Sempre cache() quando reusar um DataFrame caro: df_clean.cache() antes de medir contagens e escrever.
partitionBy em coluna de baixa cardinalidade (data, país). NUNCA em coluna de alta cardinalidade (user_id, timestamp ms).
Use coalesce() antes de escrever se está gerando milhares de arquivos pequenos: df.coalesce(8).write....
Sempre escreva _silver_ingested_at e _silver_source_path: rastreabilidade salva debug.
Evite .collect() em DataFrames grandes → puxa tudo pro driver, mata o cluster.

Pegadinhas

mode("overwrite") apaga TUDO — em produção, prefira MERGE ou overwrite com replaceWhere.
Particionar por algo errado (ex.: trip_id) → cria milhões de pastas, performance horrível e listing custoso no ADLS.
Esquecer option("overwriteSchema", "true") quando o schema mudou → falha com “schema mismatch”.
df.count() repetido dispara reprocessamento toda vez (lazy!). Cache se vai usar mais de uma vez.
from pyspark.sql.functions import * polui o namespace (conflita com min, max, etc.). Use F. sempre.

Mini exercícios (faça antes de seguir)

Adicione tip_pct = tip_amount / NULLIF(fare_amount,0) no Silver.
Mude o partitionBy para ("pickup_date","pu_location_id") e veja o que muda em arquivos gerados.
Rode DESCRIBE HISTORY e identifique seu último commit. Faça time travel para a versão 0.
Crie um bug proposital (filtro errado), regrave, depois reverta com RESTORE TABLE delta.\…` TO VERSION AS OF 0`.

Checklist de saída

Tabela Delta em silver/domain=nyctaxi/yellow_trips/ com _delta_log/.
Consegue ler com spark.read.format("delta").load(...).
DESCRIBE HISTORY mostra pelo menos 1 commit.
Particionada por pickup_date.
Sabe explicar narrow vs wide, lazy evaluation, partitionBy, OPTIMIZE/VACUUM.
Cluster terminado.

→ M05 — Azure Data Factory: orquestrar ingestão e disparar o notebook