4、数据读取与写入：读取 CSV、JSON、Parquet、ORC 等格式，写入模式与分区

数据读写的功底，说白了就是 Spark SQL 的「基本功」。你想想看，不管多牛的分析，数据进不来、出不去，全白搭。我个人习惯，拿到一个新数据集，第一件事就是搞清楚它的格式和分区策略。今天咱们就把 CSV、JSON、Parquet、ORC 这四种主流格式，连同写入模式和分区，一次性讲透。

4.1 读取 CSV：最常用，坑也最多

CSV 是数据工程师的「家常便饭」。但我在项目中遇到过太多次因为 CSV 格式不规范导致的血案。比如字段里藏着逗号、引号没转义、文件头缺失……嗯，这里要注意。

基本读取方式很简单：

df = spark.read \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .csv("hdfs://path/to/file.csv")

但别急着用 inferSchema。为什么？因为它会额外扫描一遍数据，大文件时慢得你想哭。我建议你手动指定 schema：

from pyspark.sql.types import StructType, StructField, StringType, IntegerType

schema = StructType([
    StructField("name", StringType(), True),
    StructField("age", IntegerType(), True),
    StructField("city", StringType(), True)
])

df = spark.read \
    .option("header", "true") \
    .schema(schema) \
    .csv("hdfs://path/to/file.csv")
避坑指南：我曾经接手过一个项目，CSV 文件里某个字段的值是 "N/A"，结果 inferSchema 把它推断成了 StringType，但其他行都是 IntegerType。最后 join 的时候类型不匹配，排查了整整一下午。所以，手动指定 schema 是成熟工程师的习惯。

4.2 读取 JSON：灵活但别滥用

JSON 格式灵活，嵌套结构天然适合半结构化数据。但灵活的另一面是「混乱」。同一个字段，这条记录是字符串，下条记录是数组，Spark 会直接报错。

读取 JSON 的标准写法：

df = spark.read \
    .option("multiLine", "true") \
    .json("hdfs://path/to/file.json")

这里有个小细节：multiLine 参数。如果你的 JSON 文件是每行一个 JSON 对象（即 JSON Lines 格式），就别开这个选项，否则性能会下降。如果是标准的 JSON 数组格式（整个文件一个对象），那就必须开。

个人经验：我一般优先用 JSON Lines 格式。为什么？因为可以逐行读取，支持并行处理，而且出错时只影响单行。标准 JSON 文件一旦中间格式错误，整个文件都读不了。

4.3 读取 Parquet：Spark 的「亲儿子」

Parquet 是列式存储格式，压缩率高、查询快。说白了，Spark SQL 对 Parquet 的支持是最好的。我几乎所有的生产环境数据都存成 Parquet。

读取 Parquet 最简单：

df = spark.read.parquet("hdfs://path/to/file.parquet")

不需要指定 schema，因为 Parquet 文件自带元数据。而且 Spark 会自动利用列式存储的优势，只读取你需要的列。比如你只查两列，它不会把其他列的数据也加载进来。

核心优势：Parquet 支持谓词下推（Predicate Pushdown）。你写 df.filter(col("age") > 30) 时，Spark 会在读取文件时就过滤掉不符合条件的数据行，而不是全量读进来再过滤。这在处理 TB 级数据时，性能差距是数量级的。

4.4 读取 ORC：Hive 生态的「老搭档」

ORC 和 Parquet 很像，也是列式存储。如果你在 Hive 生态里工作，ORC 可能是更自然的选择。我记得有一次帮客户迁移数据仓库，他们所有 Hive 表都是 ORC 格式，用 Spark 读起来毫无压力。

df = spark.read.orc("hdfs://path/to/file.orc")

ORC 的压缩率通常比 Parquet 更高，但查询性能略逊一筹。怎么选？我的建议是：如果你主要用 Spark，选 Parquet；如果你主要用 Hive，选 ORC。别在两个格式之间反复横跳，维护成本太高。

4.5 写入模式：覆盖、追加、忽略、报错

写入数据时，最怕的就是「写错了」。Spark 提供了四种写入模式，你想想看，选错了会出大事。

模式 含义 使用场景
overwrite 覆盖已有数据 全量刷新，比如每日全量快照
append 追加到已有数据 增量写入，比如日志流
ignore 如果目标路径已存在，直接跳过 幂等写入，防止重复跑任务
error（默认） 如果目标路径已存在，直接报错 防止误覆盖，开发环境常用

写入代码示例：

df.write \
    .mode("overwrite") \
    .parquet("hdfs://path/to/output")
避坑指南：我曾经在生产环境上犯过一个低级错误——用 overwrite 模式写数据，结果路径写错了，把上一层的整个目录给覆盖了。从那以后，我养成了一个习惯：写数据前，先用 spark._jvm.org.apache.hadoop.fs.Path 检查目标路径是否存在，确认无误再执行写入。

4.6 分区写入：让查询快 10 倍的关键

分区是数据仓库的「灵魂」。你想想看，一张 100TB 的表，如果不分区，每次查询都要全表扫描，谁也扛不住。

Spark 支持按列分区写入：

df.write \
    .partitionBy("year", "month") \
    .mode("append") \
    .parquet("hdfs://path/to/output")

这样写出来的目录结构是：

output/
  year=2024/
    month=01/
      part-00001.parquet
      part-00002.parquet
    month=02/
      part-00001.parquet
  year=2025/
    month=01/
      part-00001.parquet

查询时，如果你只查 2025 年 1 月的数据，Spark 只会读取 year=2025/month=01/ 目录下的文件，其他目录根本不会碰。这就是分区裁剪（Partition Pruning）。

个人经验：分区列的选择很有讲究。我一般选「基数适中」的列。比如按天分区，每天的数据量在 100MB 到 1GB 之间比较合适。如果每天只有几 KB，分区数太多，小文件问题会让你头疼；如果每天几十 GB，分区太大，查询时还是得扫很多数据。

4.7 动态分区与静态分区

Spark 支持两种分区写入方式：

  • 静态分区：你在代码里写死分区值，比如 .partitionBy("year", "month")，然后数据里必须有这两列。
  • 动态分区：Spark 根据数据内容自动决定数据写到哪个分区。需要开启配置：
spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

df.write \
    .mode("overwrite") \
    .partitionBy("year", "month") \
    .parquet("hdfs://path/to/output")

动态分区的好处是灵活，但坏处是容易产生大量小文件。我记得有一次，一个同事用动态分区写数据，结果生成了几万个分区目录，每个目录里只有几 KB 数据。后续读取时，光打开文件列表就花了 10 分钟。

核心建议：能用静态分区就用静态分区。如果必须用动态分区，记得控制分区数量，或者事后用 coalesce() 或 repartition() 合并小文件。

4.8 总结：选对格式，用对模式

好了，咱们把数据读写的核心点都过了一遍。总结一下我的个人经验：

  • CSV：适合小数据量、临时分析。生产环境尽量别用。
  • JSON：适合半结构化数据、日志流。注意用 JSON Lines 格式。
  • Parquet：Spark 生态的首选。列式存储、谓词下推、压缩率高。
  • ORC：Hive 生态的首选。压缩率更高，但查询性能略低。
  • 写入模式：append 用于增量，overwrite 用于全量，ignore 用于幂等。
  • 分区：按基数适中的列分区，避免小文件问题。

数据读写看似基础，但往往是整个数据 pipeline 的瓶颈。你想想看，如果读数据慢了 10 分钟，写数据又慢了 10 分钟，一天跑一次的任务，一个月就多浪费 10 个小时。把这些细节抠好，才是资深工程师该有的样子。

← 上一章 📖 返回目录 下一章 →