PySpark 读写是什么
PySpark 把 Spark 的分布式计算能力带进 Python：用 DataFrame API 读列存文件、做过滤聚合，再按分区写出结果表，也可以在同一个会话里混用 SQL。它的价值在于把 ETL 逻辑写成一段可调试的脚本，配合调度器就能成为离线管道的执行体。底层仍是 JVM 引擎，Python 侧通过 py4j 通信，因此自定义 UDF 的开销要高于内置函数。
基本信息
| 执行入口 | SparkSession.builder.appName(...).getOrCreate() |
|---|---|
| 读取接口 | spark.read.parquet / csv / orc / jdbc |
| 写出接口 | df.write.mode(...).partitionBy('dt').parquet(path) |
| 写入模式 | append 追加、overwrite 覆盖、errorifexists、ignore |
| 环境要求 | 与集群一致的 Spark 版本、匹配的 Python 版本 |
按渠道汇总示例
按渠道汇总示例
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
spark = SparkSession.builder.appName('orders_etl').getOrCreate()
df = spark.read.parquet('s3a://lake/curated/dwd/orders/dt=2026-10-11')
result = (df.filter('amount > 0') # 先过滤，减少后续扫描与聚合量
.withColumn('dt', F.lit('2026-10-11')) # 分区列要显式补上
.groupBy('channel')
.agg(F.sum('amount').alias('amount_sum'),
F.countDistinct('order_id').alias('order_cnt')))
result.write.mode('overwrite').partitionBy('dt') \
.parquet('s3a://lake/marts/ads/orders_channel_daily') # 目标必须与上游隔离典型使用场景
- 每日离线 ETL：读上游明细分区、清洗聚合后写入集市表
- 数据探查：临时读一批数据看分布、找空值与异常值，不再依赖抽样脚本
- 表格式转换：Parquet 转 Iceberg、CSV 转 ORC 这类格式统一工作
- 回补历史：按日期循环重跑指定分区，覆盖旧结果
- 小规模特征计算：直接用 DataFrame 做 join 与窗口聚合
常见参数读法
| 项 | 作用 | 注意 |
|---|---|---|
| mode | 控制目标路径已存在时的行为 | overwrite 会删除整个目标目录 |
| partitionBy | 按列写成分区目录 | 列必须已存在于 DataFrame 中 |
| repartition/coalesce | 调整写出文件数 | repartition 触发 shuffle，coalesce 不触发 |
| spark.sql.shuffle.partitions | 聚合阶段的分区数 | 直接决定输出文件数量 |
| 缓存级别 | 复用中间结果避免重复计算 | 只缓存在多分支复用时才划算 |
overwrite 会把目标路径下的内容整体替换；如果目标写成与上游共享的目录，等于把上游数据一起删掉。不确定时先写临时目录，再由调度做切换。
Python UDF 需要逐行跨进程序列化，性能明显低于内置函数；能用 functions 里的表达式解决就不要写 UDF，必须写时优先考虑 pandas UDF。