📚 课程目录第3章多格式数据源解析

3. 多格式数据源解析：常见数据格式的特点、适用场景与读写方式

3.1 核心原理：为什么湖仓需要面对“多格式”现实

湖仓一体架构的核心承诺是“一份数据，多种引擎”。但在实际落地中，数据源往往以不同格式散落在对象存储或分布式文件系统中。Parquet、ORC、Avro、JSON、CSV 这五种格式，分别代表了列式存储（Parquet、ORC）、行式序列化（Avro）、半结构化文本（JSON）和纯文本表格（CSV）四种不同的设计哲学。理解它们的差异，是选择解析策略、优化查询性能、保证数据一致性的前提。

从底层原理看，格式差异主要体现在三个维度：存储布局（行式 vs 列式）、Schema 绑定方式（强类型 vs 弱类型）、压缩编码策略（列级压缩 vs 整块压缩）。列式格式将同一列的数据连续存放，配合字典编码、位图索引和谓词下推，能大幅减少 I/O；而行式格式更适合逐条读取和流式写入。Schema 绑定方式则决定了数据写入时是否需要预先定义结构，以及演进时是否兼容新增字段。

3.2 五种格式的深度对比与选型方法

在实际项目中，选错格式往往导致查询性能下降数倍甚至数十倍。下表从存储效率、计算适配、Schema 灵活性、生态支持四个维度进行对比，帮助你在不同场景下做出合理选择。

维度	Parquet	ORC	Avro	JSON	CSV
存储布局	列式	列式	行式	文本嵌套	纯文本行式
压缩比	高（列级编码）	极高（轻量索引）	中（整块压缩）	低（文本冗余）	低（无内建压缩）
Schema 演进	支持追加列	支持追加列	原生支持（默认值）	天然灵活	无 Schema
谓词下推	优秀（统计信息）	优秀（索引+统计）	不支持	需全量扫描	需全量扫描
典型写入模式	批量写入	批量写入	流式写入	日志追加	人工导出/导入
主要引擎	Spark、Presto	Hive、Spark	Kafka、Flink	MongoDB、API	Excel、R、Python

选型决策路径：如果数据面向分析型查询（如聚合、过滤、报表），优先选择 Parquet 或 ORC；如果数据是流式管道中的中间结果，需要逐条处理且 Schema 可能变化，Avro 是更稳妥的选择；如果数据来自外部 API 或日志，且结构不固定，JSON 提供了最大的灵活性；而 CSV 仅适合小规模数据交换或人工检查，不建议作为湖仓核心存储格式。

3.3 读写方式与代码实践

在湖仓环境中，读写不同格式通常通过 Spark SQL 或 DataFrame API 完成。以下示例展示了使用 PySpark 读取不同格式并写入 Iceberg 表的核心代码。

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("MultiFormatIngest") \
    .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") \
    .config("spark.sql.catalog.demo", "org.apache.iceberg.spark.SparkCatalog") \
    .config("spark.sql.catalog.demo.type", "hadoop") \
    .config("spark.sql.catalog.demo.warehouse", "s3://data-lake/warehouse") \
    .getOrCreate()

# 1. 读取 Parquet（列式，适合分析）
df_parquet = spark.read.parquet("s3://raw-data/events/parquet/")
df_parquet.writeTo("demo.db.events").using("iceberg").create()

# 2. 读取 ORC（Hive 生态常用）
df_orc = spark.read.orc("s3://raw-data/orders/orc/")
df_orc.writeTo("demo.db.orders").using("iceberg").create()

# 3. 读取 Avro（需指定 schema）
df_avro = spark.read.format("avro") \
    .option("avroSchema", open("user.avsc").read()) \
    .load("s3://raw-data/users/avro/")

# 4. 读取 JSON（自动推断 schema，注意嵌套结构）
df_json = spark.read.option("multiline", "true") \
    .json("s3://raw-data/logs/json/")

# 5. 读取 CSV（需显式指定 schema 避免类型推断错误）
from pyspark.sql.types import StructType, StructField, StringType, TimestampType
csv_schema = StructType([
    StructField("device_id", StringType(), True),
    StructField("event_time", TimestampType(), True)
])
df_csv = spark.read.option("header", "true") \
    .schema(csv_schema) \
    .csv("s3://raw-data/legacy/device_logs.csv")

写入时需注意：Parquet 和 ORC 适合批量覆盖写入，而 Avro 更适合通过 Kafka Connect 或 Flink 以流式方式写入。对于 JSON 和 CSV，建议先转换为 DataFrame 并指定目标表 Schema，避免将弱类型问题带入湖仓。

3.4 案例：实时同步管道中的格式转换策略

某电商平台需要将 MySQL Binlog 变更、前端埋点日志、第三方广告回传文件统一汇入湖仓。原始数据格式混杂，处理管道设计如下：

MySQL Binlog → Debezium 捕获 → 序列化为 Avro 消息发送至 Kafka（利用 Schema Registry 管理演进）；
前端埋点日志 → Flume 采集 → 以 JSON 格式写入对象存储临时目录（保留原始嵌套结构）；
第三方广告回传 → 每日 FTP 推送 → CSV 文件落盘到指定桶；
Flink 作业分别消费 Kafka Avro 消息和 JSON 文件，进行清洗后以 Parquet 列式格式写入 Iceberg 表；
CSV 文件通过 Spark 批量任务读取，转换为 ORC 格式写入离线分析表。

该方案的核心收益在于：流式链路使用 Avro 保证 Schema 兼容性，分析链路使用 Parquet/ORC 提升查询性能，而 JSON 和 CSV 仅作为临时存储或外部交换格式，不进入核心分析路径。

实践建议：在湖仓中建立“格式分层”策略——原始数据区（Raw Zone）保留源格式（JSON/CSV/Avro），统一格式区（Refined Zone）转换为 Parquet 或 ORC，并注册到 Iceberg/Hudi 表。这样既保留原始可追溯性，又保证分析性能。

3.5 常见误区与复盘

误区一：所有数据都转成 Parquet 就万事大吉。Parquet 不适合频繁单行更新的场景，也不适合小文件过多的情况（列式格式的元数据开销在小文件下占比过高）。若数据是流式高频写入且需要更新，应考虑 Avro + Upsert 或直接使用 Hudi/Mor 表。

误区二：忽略 Schema 兼容性设计。某团队直接将 JSON 日志写入 Parquet 表，当业务方新增字段后，旧分区无法读取新字段，导致下游任务失败。复盘发现，应在写入前使用 Schema Registry 统一管理 Avro/Parquet 的 Schema 版本，并设置字段默认值。

误区三：CSV 类型推断导致数据静默损坏。某数据管道使用 Spark 自动推断 CSV 列类型，当某列前 1000 行全为整数、第 1001 行出现小数时，Spark 将整列推断为整数类型，小数部分被截断。复盘结论：CSV 必须显式指定 Schema，或先转换为 JSON 再推断。

误区四：压缩算法选择不当。Parquet 默认 Snappy 压缩虽然速度快，但压缩比不如 Zstd。对于冷数据存储，使用 Zstd 可节省 30% 以上空间；对于热数据查询，Snappy 的解压速度更优。应根据数据访问频率分层选择压缩策略。

复盘总结：多格式解析的核心不是“哪种格式最好”，而是在正确的位置使用正确的格式。流式管道重视 Schema 演进和序列化性能，选择 Avro；分析仓库重视扫描效率和压缩比，选择 Parquet/ORC；外部交换重视通用性，选择 JSON/CSV。同时，必须建立格式转换的标准化流程，避免因格式误用导致的数据质量问题。