数据源接入：HDFS、Hive、Kafka 的特点与接入方式
离线特征生产的第一步，就是把数据从源头搬过来。我见过不少团队，特征逻辑写得漂亮，结果卡在数据接入这一步——要么读不动，要么格式对不上，要么延迟太高。说白了，数据源选型和接入方式，决定了你整个流水线的下限。
这一章，我们聚焦三种最常用的离线数据源：HDFS、Hive、Kafka。我会结合自己的踩坑经历，把每种数据源的特点、适用场景、接入代码和注意事项都讲透。
核心观点：没有最好的数据源，只有最合适的接入方式。选型时重点考虑：数据量级、时效性要求、Schema 稳定性、下游消费能力。
一、HDFS：离线特征的基石
HDFS 是我用得最多的数据源，没有之一。为什么？因为离线特征生产本质上就是大规模批处理，而 HDFS 天生就是为这个场景设计的。
1.1 HDFS 的核心特点
- 海量存储，成本低： 我管过一个 PB 级的特征存储集群，用的就是普通服务器。HDFS 把文件切成块（默认 128MB/256MB），分散存储，还能自动冗余备份。
- 一次写入，多次读取： 文件写入后不可修改，这恰恰是离线特征需要的——保证数据一致性，不会出现读一半数据被改掉的情况。
- 适合大文件： 小文件多了 NameNode 会扛不住。我曾经遇到过因为上游日志切得太碎，导致 HDFS 上几千万个小文件，NameNode 内存直接爆掉。后来强制合并，才稳住。
1.2 HDFS 接入方式
在 Spark 中读取 HDFS 数据，最常用的是直接指定路径。我个人习惯用通配符来匹配多分区：
// Scala 示例：Spark 读取 HDFS 上的 Parquet 文件
val df = spark.read
.option("basePath", "/user/feature/raw/")
.parquet("/user/feature/raw/dt=2024-01-01/*.parquet")
// 也可以读文本文件
val textDF = spark.read
.text("/user/feature/logs/2024/01/01/part-*")
小技巧： 读取 HDFS 时，尽量指定 Schema，不要用 inferSchema。我遇到过 inferSchema 扫描大量小文件，耗时比实际处理还长。手动指定 Schema 能省 30% 以上的时间。
1.3 避坑指南
我曾经踩过的坑： 有一次读取 HDFS 上的 CSV 文件，没注意文件编码是 GBK，结果中文特征全部乱码。后来我养成了习惯：所有文本文件统一用 UTF-8 编码，并在读取时显式指定 option("encoding", "UTF-8")。
二、Hive：结构化特征的仓库
Hive 本质上是 HDFS 上的 SQL 层。你想想看，如果团队里算法同学不会写 Spark，但会写 SQL，那 Hive 就是最好的桥梁。
2.1 Hive 的核心特点
- SQL 化查询： 降低使用门槛。我团队里的数据分析师，用 HiveQL 就能直接拉取特征数据，不需要我帮忙写代码。
- 分区与分桶： 按日期、地域等维度分区，查询时只扫描相关分区，效率极高。分桶则适合做 JOIN 优化。
- Schema 强约束： 表结构固定，字段类型明确。这既是优点也是缺点——改 Schema 需要 ALTER TABLE，生产环境要谨慎。
2.2 Hive 接入方式
Spark 读取 Hive 表有两种主流方式：
| 方式 | 说明 | 适用场景 |
|---|---|---|
| SparkSQL 直接读表 | 通过 spark.sql("SELECT * FROM db.table") |
表结构稳定，需要 SQL 灵活过滤 |
| Hive Metastore 直连 | 通过 JDBC 连接 Metastore，获取表元数据 | 需要跨集群访问，或自定义分区策略 |
// SparkSQL 读取 Hive 表示例
val df = spark.sql("""
SELECT
user_id,
feature_a,
feature_b
FROM feature_db.user_features
WHERE dt = '2024-01-01'
AND feature_a IS NOT NULL
""")
注意： 使用 SparkSQL 读 Hive 表时，一定要开启动态分区裁剪：spark.sql.optimizer.dynamicPartitionPruning.enabled=true。我实测过，开启后查询速度能提升 2-5 倍。
2.3 避坑指南
我曾经踩过的坑： 有一次 Hive 表的小文件特别多（每个分区几千个小文件），Spark 读取时 task 数量爆炸，调度开销比计算还大。后来我加了 spark.sql.shuffle.partitions 和 spark.sql.files.maxPartitionBytes 参数，强制合并小文件，才把性能拉回来。
三、Kafka：实时特征的入口
Kafka 在离线特征生产中的角色比较特殊。你可能觉得奇怪：离线特征为什么要接 Kafka？其实很多场景下，我们需要把实时流数据「落地」成离线特征。比如用户实时点击日志，先进 Kafka，再通过批处理写入 HDFS 做离线训练。
3.1 Kafka 的核心特点
- 高吞吐、低延迟： 我见过单集群日处理百亿条消息的场景，Kafka 扛得住。
- 消息持久化： 消息写入磁盘，可以配置保留时间（比如 7 天）。这意味着离线任务可以回溯消费。
- 分区并行： 每个分区可以被一个消费者线程处理，天然适合分布式消费。
3.2 Kafka 接入方式
Spark Streaming 读取 Kafka 是最常见的做法。我一般用 assign 模式指定分区和 offset，避免自动提交导致丢数据：
// Spark Structured Streaming 读取 Kafka
val df = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "kafka01:9092,kafka02:9092")
.option("subscribe", "user_click_topic")
.option("startingOffsets", "earliest")
.option("failOnDataLoss", "false")
.load()
// 解析 value 为 JSON
val parsedDF = df.selectExpr("CAST(value AS STRING) as json")
.select(from_json($"json", schema).as("data"))
.select("data.*")
小技巧： 离线任务消费 Kafka 时，建议用 startingOffsets 指定 earliest，这样即使任务重启也不会丢数据。但要注意，如果消息积压太多，首次消费会很慢。我一般会先估算一下积压量，必要时提前扩容分区。
3.3 避坑指南
我曾经踩过的坑： 有一次 Kafka 集群的某个 broker 挂了，导致部分分区不可用。我的 Spark 任务没有设置 failOnDataLoss=false，结果任务直接报错退出。后来我改成 false，并加了重试机制，才保证稳定性。另外，一定要监控 consumer lag，否则离线特征会越来越滞后。
四、三种数据源的对比与选型
说了这么多，到底怎么选？我整理了一个对比表，方便你快速决策：
| 维度 | HDFS | Hive | Kafka |
|---|---|---|---|
| 数据形态 | 文件（Parquet/ORC/CSV） | 表（结构化） | 消息流（JSON/Avro） |
| 时效性 | T+1 或小时级 | T+1 或小时级 | 秒级/毫秒级 |
| 查询方式 | 路径 + 文件格式 | SQL | Consumer API |
| Schema 管理 | 弱（文件自带） | 强（Metastore） | 弱（需自行解析） |
| 典型场景 | 原始日志、模型文件 | 用户画像、统计特征 | 实时点击、行为序列 |
| 接入复杂度 | 低 | 中 | 高（需处理 offset） |
我的选型建议：
- 如果数据量大、格式固定、不需要频繁查询 → HDFS
- 如果需要 SQL 分析、团队有 SQL 基础 → Hive
- 如果需要实时接入、后续做流批一体 → Kafka
五、小结
数据源接入是离线特征生产流水线的第一道关卡。HDFS 适合海量文件存储，Hive 适合结构化查询，Kafka 适合实时流接入。选型时不要盲目追求新技术，而是看你的数据特征和团队能力。