pyspark-etl

PySpark ETL

本技能涵盖使用 PySpark、Spark SQL 和 Apache Iceberg 构建生产级、可测试 ETL 流水线的模式，包括项目结构、连接和窗口函数惯用写法，以及安全的累积表合并模式。

构建 PySpark ETL 作业的工作流

1. 搭建作业类框架 — 创建一个管理 SparkSession 生命周期的类，支持注入会话以便测试，并提供抽象的 run_job 方法。
2. 通过工厂函数定义配置 — 将配置保留为纯数据类；在单独的工厂函数中解析 CLI 参数，这样测试无需接触 sys.argv 即可构造配置。
3. 使用共享的分区感知读取器读取源数据 — 使用通用读取工具处理日期筛选、小时范围和最新分区查找；将业务筛选保留在 ETL 类中。
4. 使用 .transform() 组合流水线 — 链接命名方法（read_source().transform(self.enrich).transform(self.merge_with_existing)），让 run_job 只负责流程编排。
5. 以惯用方式应用转换 — 使用 select 替代 withColumn 链，明确指定连接类型和窗口框架，并优先使用原生函数而非 UDF。
6. 以保障架构演进安全的方式写入 — 写入 Iceberg 表时使用 .byName()，这样列顺序便无关紧要。
7. 验证输出 — 每次写入后，检查主键唯一性以及关键列的空值数量。
8. 本地测试 — 使用本地 SparkSession，通过小型手工构建的 DataFrame 对转换方法进行单元测试。

项目结构

ETL 类框架

from abc import ABC, abstractmethod
import logging
from pyspark.sql import SparkSession

class BaseETL(ABC):
    def __init__(self, config, app_name="ETL Job", spark_session=None):
        self.spark = spark_session or SparkSession.builder.appName(app_name).getOrCreate()
        self.config = config
        self.logger = logging.getLogger(self.__class__.__name__)

    @abstractmethod
    def run_job(self): ...

    def stop(self):
        self.spark.stop()

将配置定义为工厂函数

将数据类保留为纯数据；把 CLI 解析放入独立的工厂函数，以便在测试中轻松构造配置。

import argparse
from dataclasses import dataclass

@dataclass
class MyConfig:
    read_date: int = 20260101

def create_config() -> MyConfig:
    parser = argparse.ArgumentParser()
    parser.add_argument("--read_date", type=int, default=20260101)
    args = parser.parse_args()
    return MyConfig(read_date=args.read_date)

分区感知的共享读取器

构建一个用于处理分区机制的通用读取器；将领域特定的筛选条件清晰地放在 ETL 中，而不要藏在一次性读取器类里。

import pyspark.sql.functions as F

class PartitionedReader:
    @staticmethod
    def read_latest(spark, table_name, partition_col):
        row = spark.read.table(table_name).agg(F.max(partition_col)).first()
        if row is None or row[0] is None:
            return spark.createDataFrame([], spark.read.table(table_name).schema)
        return spark.read.table(table_name).filter(F.col(partition_col) == row[0])

    @staticmethod
    def read_by_date(spark, table_name, partition_col, date_value):
        return spark.read.table(table_name).filter(F.col(partition_col) == date_value)

events = PartitionedReader.read_by_date(spark, "catalog.my_table", "event_date", 20260319)
events = events.filter(F.col("event_type").isin("login", "purchase"))

代码风格

• 将函数导入为 import pyspark.sql.functions as F，并始终使用 F.col()，而不要通过 df.colA 属性访问列——属性访问会将列绑定到特定的 DataFrame 变量，在连接或重新赋值后会失效。
• 当 .filter() 或 F.when() 中的复杂布尔逻辑超过 3 个表达式时，将其提取为命名变量。
• 优先使用 select，而不是连续调用 withColumn——select 一次性声明输出架构，而每次调用 withColumn 都会向查询计划添加一个投影。
• 使用 .alias()，而不是 withColumnRenamed。
• 将每条语句中的链式方法调用限制在 5 个以内；按操作类型分开编写 select/filter 链、withColumn 链和 join 链。

# BAD — 3 intermediate DataFrames, one projection per call
df = df.withColumn("a", F.col("a").cast("double"))
df = df.withColumn("b", F.upper(F.col("b")))
df = df.withColumn("c", F.lit(1))

# GOOD — one DataFrame, explicit schema contract
df = df.select(
    F.col("a").cast("double"),
    F.upper(F.col("b")).alias("b"),
    F.lit(1).alias("c"),
)

连接

• 始终明确指定 how=，不要依赖默认值。
• 为提高可读性，优先使用左连接而不是右连接；调整 DataFrame 顺序，而不要使用 how="right"。
• 连接后，为消除歧义应为整个 DataFrame 设置别名，而不是逐列使用 withColumnRenamed 重命名。
• 使用大型事实表进行连接时，若维度表或查找表较小，则用 F.broadcast() 包装；尤其是在筛选或转换导致 Spark 无法自动推断其大小时（spark.sql.autoBroadcastJoinThreshold 只会自动广播 Spark 能确定大小的表，通常小于 10MB）。使用 df.explain() 确认执行计划——查看是否为 BroadcastHashJoin 或 SortMergeJoin。
• 不要试图用 .dropDuplicates() 掩盖意外的重复行——应找出根本原因；它还会增加 shuffle 开销。

flights = flights.alias("f")
parking = parking.alias("p")
result = flights.join(F.broadcast(parking), "code", how="left").select(
    F.col("f.start_time").alias("flight_start"),
    F.col("p.total_time").alias("parking_total"),
)

窗口函数

使用 from pyspark.sql import Window as W。

• 始终指定显式框架——如果没有指定，F.sum().over(w) 的行为会因是否存在 orderBy 而异（运行总和或总和）。
• 了解 row_number() + filter（删除行，只保留最佳行）与窗口上的 first()（覆盖一列，保留所有行）之间的区别。
• 向 F.first()/F.last() 传入 ignorenulls=True——否则，分区第一行中的 null 会使整个分区的结果都变为 null。
• 避免使用空的 partitionBy()；它会将所有数据放入单个分区。全局聚合请改用 .agg()。

w = W.partitionBy("key").orderBy("num").rowsBetween(W.unboundedPreceding, W.unboundedFollowing)
df = df.withColumn("version", F.first("version", ignorenulls=True).over(w))

Map 和数组高阶函数

• 合并 map 时，如果需要按键解决冲突（例如保留时间戳较晚的条目），而不是无条件保留一侧，请使用 map_zip_with，而非 map_concat。
• 使用 transform + array_max/array_min，无需 UDF 即可从嵌套结构体中提取值。
• 避免使用 UDF——先检查是否有内置 Spark 函数或高阶函数可用。UDF 会破坏 Catalyst 优化，并增加序列化开销。

merged = F.map_zip_with(
    new_map, existing_map,
    lambda key, v1, v2: (
        F.when(v1.isNull(), v2)
        .when(v2.isNull(), v1)
        .otherwise(F.when(v1.event_ts >= v2.event_ts, v1).otherwise(v2))
    ),
)

累积表 / 快照表模式

• 合并必须具有幂等性——使用相同输入数据重新运行不得产生重复项。
• 合并必须与顺序无关——回填旧数据绝不能覆盖较新的数据。使用明确的条件解决冲突（事件时间戳、版本号、分区日期），切勿依赖 coalesce 参数顺序这类位置优先级。
• 每次写入后，都要将主键唯一性和键列 null 数量验证作为审计步骤。

数据质量与性能

• 空列请使用 F.lit(None)——切勿使用空字符串或 "NA" 之类的哨兵值。
• 对类别映射的 F.when() 链，避免使用 .otherwise() 兜底；未映射的值应显示为 null，而不是被悄悄归为“其他”。
• 生产代码中绝不能保留 .show()、.collect() 或 .printSchema()——它们会强制完整物化或增加 driver 开销。若有意用于行数日志记录，或在 DAG 分叉前强制物化，则可以使用 .count()。
• 仅当后续多个操作会引用同一个 DataFrame 时才使用 .persist()。有意选择存储级别：MEMORY_AND_DISK（安全的默认选项）、MEMORY_ONLY（最快，但驱逐时有重新计算的风险）、DISK_ONLY（适用于大到无法放入内存的 DataFrame）。

Iceberg 写入模式

• 写入时使用 .byName()，让 Spark 按名称而非位置匹配列——这样可确保写入在架构演进期间依然安全。

df.write.byName().mode("overwrite").insertInto("catalog.my_table")

• 使用 __partitions Iceberg 元数据表查找最新快照，而不是扫描整张表：

partition_df = spark.read.table("catalog.my_table__partitions").select(
    "partition.partition_date", "partition.partition_hour"
)
max_partition = partition_df.orderBy(
    F.col("partition_date").desc(), F.col("partition_hour").desc()
).first()
if max_partition is None:
    raise ValueError("No partitions found in catalog.my_table")

• 有意选择 write.distribution-mode："none"（最快，不重新分区，文件大小取决于上游分区方式）、"hash"（按分区键重新分区，以生成大小均匀的文件）、"range"（写入前排序，扫描性能最佳，但开销最高）。