4. 数据摄入机制
数据摄入，说白了就是数据怎么进 Pinot 的。这是整个系统的入口，也是很多线上问题的源头。我见过不少团队，表建得漂漂亮亮，查询也优化得不错，结果卡在数据进不来、进得慢、进得不一致上。
Pinot 提供了两套完全独立的摄入引擎：离线批量和实时流式。它们各自解决不同场景的问题，但底层又共享同一套存储和索引结构。今天我就把这两套引擎掰开揉碎了讲清楚。
核心观点：离线引擎负责历史数据的高吞吐导入，实时引擎负责增量数据的低延迟摄入。两者通过 Segment 生命周期管理无缝衔接，构成完整的时序数据闭环。
4.1 离线批量导入：Hadoop/Spark 生态的深度整合
离线导入，就是批处理。你想想看，每天凌晨跑一次 ETL，把昨天的数据从 Hive 或者 HDFS 上拉下来，转成 Pinot 的 Segment 格式，然后推送到 Pinot 集群。这是最经典的做法。
Pinot 的离线导入依赖一个核心组件——PinotSegmentGenerator。它本质上是一个 MapReduce 或 Spark 作业，负责把原始数据转换成列式存储的 Segment 文件。
4.1.1 离线导入的完整流程
- 数据准备：原始数据存储在 HDFS、Hive 表或 S3 上，通常是 Parquet、Avro 或 CSV 格式。
- Segment 生成：通过 Hadoop/Spark 作业读取数据，按 Pinot 的列式格式重新编码，生成 Segment 文件。
- Segment 上传：生成的 Segment 文件上传到 Pinot Controller 指定的深存储（如 HDFS、S3）。
- Segment 推送：Controller 收到上传通知后，将 Segment 分配给合适的 Server 节点加载。
- Segment 加载：Server 从深存储拉取 Segment，加载到内存，开始服务查询。
我的经验：我曾经遇到一个坑——Segment 生成时字典编码的配置没调好。默认的字典编码对高基数列（比如用户 ID）会占用大量内存。后来我强制对这类列关闭了字典编码，Segment 大小直接降了 40%。
4.1.2 离线导入的配置示例
下面是一个典型的 Spark 作业配置，用于生成离线 Segment：
// Spark 作业配置示例
val jobConfig = new PinotSparkJobConfig()
.setTableName("myTable_OFFLINE")
.setInputDataDir("hdfs://namenode:8020/data/raw/2024/01/01")
.setInputFormat("parquet")
.setOutputDir("hdfs://namenode:8020/pinot/segments/")
.setSegmentNamePrefix("myTable_20240101")
.setOverwrite(true)
.setNumSegments(10) // 控制并行度
// 启动 Segment 生成作业
PinotSparkSegmentGenerator.run(jobConfig, sparkSession)这里有个关键参数 numSegments。它决定了生成多少个 Segment 文件。我建议根据数据量和集群规模来调：每个 Segment 控制在 200MB 到 500MB 之间比较合适。太小了文件碎片多，太大了加载和查询都慢。
4.1.3 离线导入的优缺点
| 维度 | 说明 |
|---|---|
| 吞吐量 | 极高，依赖 Hadoop/Spark 的分布式计算能力，可以处理 PB 级数据 |
| 延迟 | 高，从数据产生到可查询通常有数小时到一天的延迟 |
| 数据一致性 | 强一致，Segment 生成完成后原子性替换 |
| 适用场景 | 历史数据回填、T+1 报表、全量快照 |
4.2 实时流式摄入：Kafka 的无缝对接
离线导入搞定了历史数据，但实时场景怎么办？比如用户点击流、交易流水、监控指标——这些数据要求秒级可见。这时候就得靠实时流式摄入。
Pinot 的实时引擎直接对接 Kafka（也支持 Kinesis 和 Pub/Sub），数据从 Topic 消费后，在内存中构建 Segment，然后定期 flush 到深存储。
4.2.1 实时摄入的核心机制
实时摄入的关键在于 Consumer 和 Segment 生命周期管理。每个实时表对应一个 Kafka Topic，Pinot 的 Server 节点作为 Consumer 消费数据。
- 实时 Segment：数据先写入内存中的 Mutable Segment，可读可写。
- Segment 完成：当 Mutable Segment 达到时间阈值或大小阈值后，转为 Immutable Segment，flush 到深存储。
- Segment 替换：Controller 将新的 Immutable Segment 分配给其他 Server 加载，同时通知原 Server 卸载 Mutable Segment。
注意：实时摄入的延迟取决于 flushThresholdTime 和 flushThresholdRows 两个参数。我建议把时间阈值设为 5 分钟，行数阈值设为 500 万行。这样既能保证数据及时可见，又不会让 Mutable Segment 占用太多内存。
4.2.2 实时表的配置示例
这是我在项目中常用的实时表配置：
{
"tableName": "myTable_REALTIME",
"tableType": "REALTIME",
"segmentsConfig": {
"replication": "3",
"timeColumnName": "eventTime",
"timeType": "MILLISECONDS",
"retentionTimeUnit": "DAYS",
"retentionTimeValue": "30"
},
"ingestionConfig": {
"streamIngestionConfig": {
"streamConfigMaps": [{
"streamType": "kafka",
"streamConfigs": {
"topicName": "myTopic",
"bootstrapServers": "kafka-broker:9092",
"consumerType": "LOWLEVEL",
"decoderClass": "org.apache.pinot.plugin.inputformat.json.JSONMessageDecoder"
}
}]
},
"segmentFlushConfig": {
"flushThresholdTime": "300000", // 5 分钟
"flushThresholdRows": "5000000" // 500 万行
}
}
}这里有个细节——consumerType 我选了 LOWLEVEL。为什么不用 HIGHLVEL？因为低级消费者可以精确控制每个 Partition 的消费进度，在数据重平衡时更稳定。高级消费者虽然方便，但遇到 Consumer Group 重平衡时容易丢数据。这是我踩过的坑，分享给你。
4.2.3 实时摄入的优缺点
| 维度 | 说明 |
|---|---|
| 吞吐量 | 中等，受限于单 Server 的内存和 CPU，但可以通过增加 Partition 扩展 |
| 延迟 | 低，秒级到分钟级，取决于 flush 阈值 |
| 数据一致性 | 最终一致，存在短暂的数据窗口期 |
| 适用场景 | 实时监控、在线分析、事件流处理 |
4.3 双引擎的协同与选择
离线引擎和实时引擎不是互斥的。实际上，Pinot 支持 Hybrid Table——同一张表同时挂载离线 Segment 和实时 Segment。查询时自动合并两部分数据，对用户透明。
我建议这样选型：
- 历史数据 + 增量数据：用 Hybrid Table，离线导入历史全量，实时摄入增量。
- 纯实时场景：只用实时表，比如监控指标。
- 纯离线场景：只用离线表，比如 T+1 报表。
避坑指南：我曾经在一个项目里把实时表的 flush 阈值设得太小（1 分钟），结果 Segment 文件数量爆炸，Controller 的元数据管理直接 OOM。后来我把阈值调到 10 分钟，问题解决。记住——实时不是越快越好，要平衡延迟和系统开销。
4.4 双引擎架构总览
下面这张图展示了离线与实时摄入的完整架构，以及它们如何汇聚到统一的查询层：
从这张图可以看得很清楚：离线引擎和实时引擎各自独立工作，但最终都产出 Segment 文件。查询时，Broker 统一路由到对应的 Server，Server 同时加载离线 Segment 和实时 Segment，对用户完全透明。
4.5 小结
离线批量导入和实时流式摄入，是 Pinot 数据摄入的两条腿。缺了离线，你没法处理历史数据；缺了实时，你没法做秒级分析。两者配合使用，才能构建一个完整的数据分析平台。
嗯，这一章的内容就到这里。记住我提到的那些参数调优经验和避坑指南，在实际部署时能帮你省不少时间。