4. ODS层建设：Kafka数据入湖（Hudi）、ODS层表设计、数据分区策略、数据压缩与存储优化

ODS层，说白了就是原始数据层。它存的是从业务系统捞过来的最原始的数据，几乎不做任何业务逻辑处理。很多刚入行的朋友会问：既然不做处理，那直接放Kafka不就行了？嗯，这里要注意——Kafka的消息有保留期限，而且不支持高效的历史查询。所以我们需要把数据从Kafka搬到湖里，也就是Hudi。

我个人习惯把ODS层比作「数据的水源地」。源头脏一点没关系，但必须完整、可回溯。这一章我们就来聊聊怎么把Kafka的数据安全地倒进Hudi，以及表怎么设计、分区怎么切、存储怎么省。

4.1 Kafka数据入湖（Hudi）

Kafka到Hudi的入湖，我经历过好几个方案。最早用Spark Streaming写Hudi，后来用Flink CDC，现在主流是Flink + Hudi Connector。为什么？因为Flink的流处理能力更强，而且Hudi的Flink Connector已经比较成熟了。

我在项目中遇到过一个问题：Kafka的Topic有几十个分区，Hudi表也有几十个分区，如果并行度设置不对，很容易出现数据倾斜。后来我总结了一套配置模板，分享给你。

4.1.1 Flink + Hudi入湖配置

-- 创建Hudi表（Flink SQL）
CREATE TABLE ods_order (
    order_id BIGINT,
    user_id BIGINT,
    product_id BIGINT,
    amount DECIMAL(10,2),
    order_ts TIMESTAMP(3),
    proc_time AS PROCTIME(),
    PRIMARY KEY(order_id) NOT ENFORCED
) WITH (
    'connector' = 'hudi',
    'path' = 'hdfs://namenode:8020/warehouse/ods/ods_order',
    'table.type' = 'COPY_ON_WRITE',
    'write.operation' = 'upsert',
    'write.precombine.field' = 'order_ts',
    'hoodie.datasource.write.recordkey.field' = 'order_id',
    'hoodie.datasource.write.partitionpath.field' = 'dt',
    'write.tasks' = '4',
    'compaction.tasks' = '1',
    'compaction.async.enabled' = 'true',
    'compaction.trigger.strategy' = 'num_commits',
    'compaction.delta_commits' = '5'
);

这里有几个关键点：

• table.type：我建议ODS层用COPY_ON_WRITE。虽然MERGE_ON_READ写入更快，但查询时需要合并，ODS层经常被下游读取，COW的查询性能更好。
• write.precombine.field：这个字段用于去重。如果同一条数据来了多次，保留时间戳最大的那条。我习惯用业务时间戳，而不是处理时间。
• compaction：异步压缩一定要开。我曾经因为没开压缩，小文件越积越多，最后NameNode压力爆表。

4.1.2 入湖性能调优

入湖性能，说白了就是「写多快」和「写多稳」。我踩过不少坑，这里列几个最实用的：

4.2 ODS层表设计

ODS层的表设计，我遵循三个原则：

1. 保持原始结构：字段名、字段类型尽量和源系统一致，不要做类型转换。
2. 添加技术字段：比如入湖时间、数据来源、数据版本等。
3. 主键设计：尽量使用业务主键，如果没有，可以用联合主键。

你想想看，如果ODS层就做了类型转换，下游排查问题时还得猜原始数据长什么样，多麻烦。

4.2.1 表结构示例

-- ODS层订单表
CREATE TABLE ods_order (
    -- 业务字段（保持原始）
    order_id BIGINT,
    user_id BIGINT,
    product_id BIGINT,
    amount DECIMAL(10,2),
    order_status INT,
    order_ts TIMESTAMP(3),
    
    -- 技术字段（Hudi自动生成）
    _hoodie_commit_time STRING,
    _hoodie_commit_seqno STRING,
    _hoodie_record_key STRING,
    _hoodie_partition_path STRING,
    _hoodie_file_name STRING,
    
    -- 自定义技术字段
    dt STRING,           -- 分区字段
    source_system STRING, -- 数据来源
    load_ts TIMESTAMP(3)  -- 入湖时间
) WITH (
    'connector' = 'hudi',
    'path' = 'hdfs://namenode:8020/warehouse/ods/ods_order',
    'table.type' = 'COPY_ON_WRITE',
    'write.operation' = 'upsert',
    'write.precombine.field' = 'order_ts',
    'hoodie.datasource.write.recordkey.field' = 'order_id',
    'hoodie.datasource.write.partitionpath.field' = 'dt'
);

4.3 数据分区策略

分区策略，说白了就是「数据怎么切」。切得好，查询快、管理方便；切得不好，小文件成灾、查询慢如蜗牛。

我见过最坑的分区策略是按小时分区。你想想看，一天24个分区，一个月720个分区。Hive Metastore里分区元数据膨胀得厉害，NameNode也扛不住。后来我统一改成按天分区。

4.3.1 分区字段选择

我个人习惯：ODS层统一用 dt 字段做分区，格式为 yyyy-MM-dd。如果业务确实需要小时级查询，可以在DWD层做小时分区，ODS层保持天分区。

4.3.2 分区裁剪优化

分区裁剪，就是查询时只扫描需要的分区。这个优化很关键，尤其是ODS层数据量大，全表扫描基本不可接受。

-- 好的写法（利用分区裁剪）
SELECT * FROM ods_order WHERE dt = '2024-01-15';

-- 坏的写法（全表扫描）
SELECT * FROM ods_order WHERE order_ts >= '2024-01-15 00:00:00';

4.4 数据压缩与存储优化

存储优化，说白了就是「省钱」。Hudi底层用Parquet格式存储，Parquet本身支持多种压缩算法。选对了压缩算法，存储空间能省一半以上。

4.4.1 压缩算法对比

我推荐ODS层用Zstd。为什么？压缩比接近Gzip，但解压速度比Gzip快很多。ODS层的数据下游要频繁读取，解压速度直接影响查询性能。

4.4.2 Hudi压缩配置

-- 设置压缩算法为Zstd
hoodie.parquet.compression.codec = zstd

-- 控制文件大小（避免小文件）
hoodie.parquet.max.file.size = 134217728  -- 128MB
hoodie.parquet.small.file.limit = 104857600  -- 100MB

-- 小文件自动合并
hoodie.clustering.inline = true
hoodie.clustering.inline.max.commits = 4
hoodie.clustering.plan.strategy.target.file.max.bytes = 134217728
hoodie.clustering.plan.strategy.small.file.limit = 104857600

4.4.3 小文件治理

小文件是HDFS的噩梦。每个小文件都会占用NameNode的内存，而且查询时MapTask数量会暴增。Hudi提供了Clustering机制来自动合并小文件。

我在项目中遇到过：ODS层一天的数据产生了上千个小文件，每个只有几MB。NameNode内存飙升到90%，差点OOM。后来我开启了Clustering，每4次提交触发一次合并，小文件数量降到了几十个。

4.5 本章小结

ODS层建设，核心就三件事：入湖、分区、压缩。入湖要稳，分区要准，压缩要狠。记住：ODS层是数据仓库的根基，根基不稳，上层再花哨也没用。

我个人习惯在ODS层多花点心思，把表结构、分区策略、压缩配置都定好。后面DWD、DWS层的建设就会顺畅很多。你想想看，如果ODS层的数据都乱七八糟，下游怎么敢用？