13 Transformación, Carga y Pipeline ETL
Nota
Continuación del capítulo anterior. Asume que tienes el DataFrame df cargado desde DBFS.
13.1 Transformación (Transform)
13.1.1 Funciones de columna (pyspark.sql.functions)
from pyspark.sql.functions import (
col, lit, when, coalesce,
to_timestamp, date_format, hour, dayofweek, month,
regexp_replace, trim, upper, lower,
round as spark_round, abs as spark_abs,
log1p, sqrt,
sum as spark_sum, avg, count, countDistinct,
lag, lead, rank, dense_rank,
window
)13.1.2 Limpieza de datos a escala
# Filtrar viajes imposibles (distancia negativa, tarifa negativa, etc.)
df_clean = (df
.filter(col("trip_distance") > 0)
.filter(col("fare_amount") > 0)
.filter(col("total_amount") > 0)
.filter(col("passenger_count").between(1, 6))
# Eliminar outliers extremos (99.9 percentil ≈ 120 USD total)
.filter(col("total_amount") < 500)
)
# Imputar nulos con valor por defecto
df_clean = df_clean.fillna({
"passenger_count": 1,
"trip_distance": 0.0,
})
# Crear columnas derivadas
df_feat = (df_clean
.withColumn("hora_recogida", hour("tpep_pickup_datetime"))
.withColumn("dia_semana", dayofweek("tpep_pickup_datetime"))
.withColumn("mes", month("tpep_pickup_datetime"))
.withColumn("duracion_min",
(col("tpep_dropoff_datetime").cast("long") -
col("tpep_pickup_datetime").cast("long")) / 60)
.withColumn("tarifa_por_km",
when(col("trip_distance") > 0,
spark_round(col("fare_amount") / col("trip_distance"), 2))
.otherwise(None))
.withColumn("propina_pct",
spark_round(col("tip_amount") / col("fare_amount") * 100, 1))
)13.1.3 Aggregaciones y Window Functions
from pyspark.sql import Window
# Agregación simple: ingresos por hora del día
ingresos_hora = (df_feat
.groupBy("hora_recogida")
.agg(
count("*").alias("n_viajes"),
avg("total_amount").alias("ticket_promedio"),
spark_sum("total_amount").alias("ingreso_total")
)
.orderBy("hora_recogida")
)
ingresos_hora.show(24)
# Window function: ranking de zonas por ingreso diario
wspec = Window.partitionBy("mes","dia_semana").orderBy(col("ingreso_total").desc())
ranking = (df_feat
.groupBy("mes","dia_semana","PULocationID")
.agg(spark_sum("total_amount").alias("ingreso_total"))
.withColumn("rank", dense_rank().over(wspec))
.filter(col("rank") <= 5)
.orderBy("mes","dia_semana","rank")
)
ranking.show(20)13.1.4 Joins a escala
# Cargar tabla de zonas (pequeña — cabe en memoria → broadcast join)
from pyspark.sql.functions import broadcast
zonas = spark.read.csv("data/taxi_zone_lookup.csv",
header=True, inferSchema=True)
# Broadcast join: Spark envía la tabla pequeña a todos los executors
df_con_zona = df_feat.join(
broadcast(zonas),
df_feat.PULocationID == zonas.LocationID,
"left"
)
# Para tablas grandes: sort-merge join (default automático en Spark)
df_join_grande = df_feat.join(otra_tabla_grande, on="clave", how="inner")
Advertencia
Evitar el Cartesian Product. Un join sin condición sobre DataFrames grandes puede generar N×M filas y agotar la memoria del clúster. Spark lanzará un error si intentas un cross join sin .crossJoin() explícito.
13.1.5 Ejercicio E-2 — Pipeline de limpieza completo
13.2 Transformaciones SQL con Spark
Spark SQL permite usar sentencias SQL estándar sobre DataFrames:
# Registrar el DataFrame como vista temporal
df_feat.createOrReplaceTempView("viajes")
# Consulta SQL — idéntica a lo que harías en BigQuery o Snowflake
resultado = spark.sql("""
SELECT
hora_recogida,
dia_semana,
COUNT(*) AS n_viajes,
ROUND(AVG(total_amount), 2) AS ticket_promedio,
ROUND(AVG(propina_pct), 1) AS propina_pct_promedio,
ROUND(SUM(total_amount), 0) AS ingreso_total
FROM viajes
WHERE total_amount BETWEEN 2.5 AND 200
GROUP BY hora_recogida, dia_semana
ORDER BY hora_recogida, dia_semana
""")
resultado.show(48)
Tip
Si ya conoces SQL, puedes empezar con Spark SQL y aprender la API DataFrame gradualmente. Ambas generan exactamente el mismo plan de ejecución optimizado.
13.3 Carga (Load)
13.3.1 Escritura en Parquet (formato estándar del lakehouse)
# Escritura simple
df_feat.write.mode("overwrite").parquet("output/viajes_limpios/")
# Particionado — clave para consultas eficientes (pruning de particiones)
df_feat.write\
.mode("overwrite")\
.partitionBy("mes", "dia_semana")\
.parquet("output/viajes_por_fecha/")
# Controlar el número de archivos de salida
df_feat.coalesce(8).write.mode("overwrite").parquet("output/viajes_8archivos/")13.3.2 Delta Lake — el formato lakehouse con ACID
# pip install delta-spark
from delta import configure_spark_with_delta_pip, DeltaTable
spark = configure_spark_with_delta_pip(
SparkSession.builder.appName("ETL-Delta")
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
.config("spark.sql.catalog.spark_catalog",
"org.apache.spark.sql.delta.catalog.DeltaCatalog")
).getOrCreate()
# Escritura inicial
df_feat.write.format("delta").mode("overwrite").save("output/delta/viajes/")
# Upsert (MERGE) — fundamental para cargas incrementales
delta_table = DeltaTable.forPath(spark, "output/delta/viajes/")
nuevos = spark.read.parquet("data/yellow_tripdata_2023-02.parquet")
(delta_table.alias("viejo")
.merge(nuevos.alias("nuevo"),
"viejo.VendorID = nuevo.VendorID AND "
"viejo.tpep_pickup_datetime = nuevo.tpep_pickup_datetime")
.whenMatchedUpdateAll()
.whenNotMatchedInsertAll()
.execute())
# Time travel — leer versión anterior de la tabla
df_ayer = spark.read.format("delta").option("versionAsOf", 0)\
.load("output/delta/viajes/")13.3.3 Escritura a una base de datos (JDBC)
(df_feat.write
.format("jdbc")
.option("url", "jdbc:postgresql://dw-host:5432/datawarehouse")
.option("dbtable", "hechos.viajes_taxi")
.option("user", "etl_user")
.option("password", "secret")
.option("batchsize", 10000) # filas por batch INSERT
.option("numPartitions", 8) # conexiones paralelas
.mode("append")
.save())13.4 Pipeline ETL completo
Ahora unimos Extract + Transform + Load en una función reutilizable:
from pyspark.sql import SparkSession, DataFrame
from pyspark.sql.functions import (
col, hour, dayofweek, month, when,
spark_round as _round
)
import logging
logging.basicConfig(level=logging.INFO)
log = logging.getLogger("ETL-Taxi")
def extract(spark: SparkSession, path: str) -> DataFrame:
log.info(f"Leyendo {path}")
return spark.read.parquet(path)
def transform(df: DataFrame) -> DataFrame:
log.info("Transformando...")
return (df
.filter(col("trip_distance") > 0)
.filter(col("fare_amount").between(2.5, 500))
.filter(col("passenger_count").between(1, 6))
.fillna({"passenger_count": 1})
.withColumn("hora", hour("tpep_pickup_datetime"))
.withColumn("dia", dayofweek("tpep_pickup_datetime"))
.withColumn("mes", month("tpep_pickup_datetime"))
.withColumn("duracion",
(col("tpep_dropoff_datetime").cast("long") -
col("tpep_pickup_datetime").cast("long")) / 60)
.filter(col("duracion").between(1, 180))
.withColumn("propina_pct",
when(col("fare_amount") > 0,
_round(col("tip_amount")/col("fare_amount")*100, 1))
.otherwise(0.0))
)
def load(df: DataFrame, output_path: str) -> None:
log.info(f"Escribiendo en {output_path}")
(df.write
.mode("overwrite")
.partitionBy("mes", "dia")
.parquet(output_path))
log.info("¡ETL completo!")
if __name__ == "__main__":
spark = (SparkSession.builder
.appName("ETL-NYC-Taxi")
.master("local[*]")
.config("spark.sql.shuffle.partitions", "16")
.getOrCreate())
raw = extract(spark, "data/yellow_tripdata_2023-01.parquet")
clean = transform(raw)
load(clean, "output/taxi_etl/")
spark.stop()