Avaliação preguiçosa: transformações e ações

Os exemplos de PySpark desta wiki precisam de um cluster Spark e não rodam aqui. O teste do site confere que cada um é Python válido (py_compile); a saída, quando aparece, é a que a documentação descreve.

Transformação monta, ação executa

from pyspark.sql import functions as F

pedidos = spark.read.table("silver.pedidos")            # nada lido ainda
entregues = pedidos.where(F.col("status") == "entregue") # ainda nada
por_dia = entregues.groupBy("data").agg(F.sum("valor").alias("total"))  # ainda nada

por_dia.count()   # ação: agora o Spark otimiza o plano inteiro e roda

Transformações (select, where, withColumn, join, groupBy().agg(), orderBy) devolvem um novo DataFrame e só acrescentam um passo ao plano. Ações (count, show, collect, take, write, toPandas) pedem um resultado e disparam um ou mais *jobs*.

Por que esperar é bom

Vendo o plano inteiro antes de rodar, o otimizador (Catalyst) consegue:

• empurrar o filtro para a leitura (*predicate pushdown*): o Parquet/Delta pula blocos inteiros;
• ler só as colunas usadas (*column pruning*);
• escolher a estratégia de join pelo tamanho estimado de cada lado.

por_dia.explain(mode="formatted")

O explain mostra o plano físico sem rodar nada. Procure PushedFilters na leitura e Exchange (shuffle) entre os estágios.

A armadilha: cada ação recalcula

Sem cache, duas ações sobre o mesmo DataFrame refazem o plano inteiro, leitura incluída.

caro = pedidos.join(clientes, "cliente_id").where("valor > 100")
caro.count()            # lê e junta
caro.write.saveAsTable("gold.pedidos_relevantes")  # lê e junta de novo

Se o resultado intermediário é reaproveitado, grave uma tabela ou use cache() (veja Cache e persist).

• collect() traz tudo para o driver: em tabela grande, derruba o driver. Use take(n) ou limit(n).
• print(df) não roda nada; mostra só o esquema. df.show() é uma ação.
• Ler o esquema pode disparar um job se ele precisar ser inferido (CSV/JSON sem esquema declarado).