1. 为什么需要双模式数据接入在数据分析领域实时流处理和离线批处理就像人的左右手各自擅长不同的场景。我遇到过不少团队刚开始只配置了Kafka实时接入结果遇到历史数据回溯时就抓瞎也有些团队只用HDFS批处理等到老板要看实时看板时只能干瞪眼。Druid的聪明之处在于它原生支持两种数据摄入模式。实时流处理Kafka适合监控报警、实时大屏这类对延迟敏感的场景数据从产生到可查询通常在秒级。而离线批处理HDFS则是数据仓库、历史分析的基石能可靠地处理TB级的历史数据。最近给一个电商客户做日志分析系统时我们就用到了这种双模式用Kafka实时监控网站异常访问同时每天用HDFS离线处理完整的用户行为日志。两种方式共用同一套数据Schema确保指标计算口径一致。2. 环境准备与前置条件2.1 基础设施检查清单在开始配置前建议先准备好以下环境Druid集群推荐Imply发行版本文基于3.0.4确保Overlord、MiddleManager等核心服务正常Hadoop集群需要确认HDFS和YARN服务可用特别是检查NameNode和ResourceManager状态Kafka集群本文使用Kafka 3.0.0需要确保Zookeeper和Broker服务正常运行验证HDFS可用性的快速方法hadoop fs -ls hdfs://your-namenode:8020/检查Kafka集群状态的命令kafka-topics.sh --bootstrap-server kafka-broker:9092 --list2.2 Druid扩展组件安装Druid需要通过扩展来支持不同数据源# 安装Kafka索引扩展 bin/load-extention --download druid-kafka-indexing-service # 安装Hadoop依赖 bin/load-extention --download druid-hdfs-storage遇到过有团队因为漏装扩展折腾半天才发现问题。特别提醒扩展版本需要与Druid核心版本严格匹配。3. HDFS离线批处理全配置解析3.1 数据准备与上传假设我们有个网站访问日志文件access.log格式如下{timestamp:2023-01-01T12:00:00Z,url:/product/123,userId:user1,region:CN}上传到HDFS的实操命令# 创建专用目录 hadoop fs -mkdir -p /druid/input # 上传测试文件 hadoop fs -put access.log /druid/input/3.2 核心配置文件拆解完整的index_hdfs.json配置包含三大模块数据模式(dataSchema){ dataSource: web_logs, parser: { type: hadoopyString, parseSpec: { format: json, dimensionsSpec: { dimensions: [url, userId, region] }, timestampSpec: { column: timestamp, format: iso } } }, metricsSpec: [ { type: count, name: views } ], granularitySpec: { segmentGranularity: DAY, queryGranularity: HOUR } }IO配置(ioConfig)ioConfig: { type: hadoop, inputSpec: { type: static, paths: /druid/input/access.log } }调优参数(tuningConfig)tuningConfig: { type: hadoop, partitionsSpec: { type: hashed, targetPartitionSize: 5000000 }, jobProperties: { mapreduce.map.memory.mb: 2048, mapreduce.reduce.memory.mb: 4096 } }踩坑提醒segmentGranularity设置过小会导致segment爆炸过大会影响查询效率。对于日活百万级的应用DAY粒度通常比较合适。4. Kafka实时流处理实战4.1 Kafka主题准备创建专用Topic的命令kafka-topics.sh --create \ --bootstrap-server kafka1:9092 \ --topic web_events \ --partitions 3 \ --replication-factor 2建议partition数量根据消费者数量调整我们一般设置为消费者数量的1.5倍。4.2 实时摄取配置详解完整的kafka_index.json配置示例{ type: kafka, dataSchema: { dataSource: web_events_realtime, parser: { type: string, parseSpec: { format: json, timestampSpec: { column: timestamp, format: iso }, dimensionsSpec: { dimensions: [url, userId, region] } } }, metricsSpec: [ { type: count, name: count }, { type: doubleSum, name: loadTime, fieldName: loadTime } ] }, ioConfig: { topic: web_events, consumerProperties: { bootstrap.servers: kafka1:9092,kafka2:9092, auto.offset.reset: earliest }, taskCount: 2, replicas: 1, taskDuration: PT10M }, tuningConfig: { maxRowsInMemory: 100000, maxBytesInMemory: 100000000 } }关键参数说明taskDuration控制任务重启间隔太短会导致频繁重启太长会影响均衡maxRowsInMemory内存中最大行数需要根据JVM堆大小调整auto.offset.reset建议从最早开始消费避免漏数据4.3 生产测试数据通过控制台生产者发送测试数据kafka-console-producer.sh --broker-list kafka1:9092 --topic web_events {timestamp:2023-01-01T12:00:01Z,url:/home,userId:user2,region:US,loadTime:1.2}5. 双模式数据一致性验证5.1 数据比对方法为确保实时和离线数据一致我们通常执行以下检查基数校验SELECT COUNT(DISTINCT userId) FROM web_logs SELECT COUNT(DISTINCT userId) FROM web_events_realtime指标对比SELECT SUM(views) FROM web_logs WHERE __time BETWEEN TIMESTAMP 2023-01-01 AND TIMESTAMP 2023-01-02 SELECT SUM(count) FROM web_events_realtime WHERE __time BETWEEN TIMESTAMP 2023-01-01 AND TIMESTAMP 2023-01-025.2 常见问题排查时间窗口不对齐检查两边配置的timestampSpec格式是否一致特别是时区设置。曾经有个项目因为实时流用了UTC时间离线用了本地时间导致数据对不上。维度值缺失确认dimensionsSpec包含所有需要的维度字段。遇到过有团队在离线配置里漏了region字段结果聚合分析时发现数据不全。6. 性能调优实战经验6.1 批处理优化技巧合理设置分区大小targetPartitionSize建议设为500-1000万行过小会导致任务数爆炸调整YARN资源jobProperties: { mapreduce.map.memory.mb: 4096, mapreduce.reduce.memory.mb: 8192 }6.2 实时流优化要点内存控制maxRowsInMemory和maxBytesInMemory需要平衡查询性能和内存压力并行度调整taskCount应该与Kafka partition数成倍数关系在最近的一个性能调优案例中通过调整segmentGranularity从HOUR到DAY使得系统吞吐量提升了3倍同时查询延迟仅增加10%。7. 运维监控方案7.1 关键指标监控建议监控以下核心指标延迟指标ingest/lagKafka消费延迟资源使用segment/usedBytes存储空间使用错误率task/failed任务失败计数7.2 自动化运维脚本定期清理旧任务的脚本示例curl -X DELETE http://druid-overlord:8081/druid/indexer/v1/task/{taskId}对于生产环境建议配置自动化的任务失败告警和自动重试机制。