尧图网站建设 
简介面向天气数据实时采集与处理场景这份项目包以完整 Java 工程形式演示了从爬虫抓取天气数据经 Kafka 实时分发再由 Flume 消费并导入 HBase随后通过 Hive 建立映射、Superset 展示分析结果的完整链路。适合大数据实时数仓、消息中间件与 NoSQL 存储方向的开发者参考。压缩包共 17 个文件含 11 个 Java 源码、1 个 XML 工程配置、3 张 PNG 架构示意图及说明文档整体约 985KB体量轻、目录清晰便于快速阅读工程结构。已有 152 人学习下载。该示例覆盖数据源接入、Topic 设计、Flume 导入、HBase 表结构及 Hive 外表映射等关键点可作为实际项目的可运行参考并二次扩展。1. 天气爬虫采集到 HBase一条实时数据管道的完整落地路径「天气爬虫采集Kafka 实时分发Flume 收集数据导入 HBase」——这串标题看着像三个工具拼在一起实际是一条完整的数据管道采集端用爬虫把天气数据拿下来经 Kafka 做实时缓冲与分发再由 Flume 落进 HBase 供查询和分析。这套组合适合两类人一类在做城市天气、气象实况类的数据服务另一类想照葫芦画瓢搭「爬虫 → 消息队列 → 列式存储」这套通用实时管道。本文按我实际落地的顺序讲采集代码怎么写、Kafka topic 和分区怎么设、Flume 配置文件怎么填以及最让我翻车的那几个坑。2. 天气爬虫采集先想清楚「采什么、发去哪」再写第一行代码2.1 数据源与采集策略为什么天气数据天然适合这条管道天气数据和爬虫常见的电商、社交数据不一样它有两个对管道设计影响很大的特点单条记录小、字段结构固定但更新频率高——温度、湿度、风力每隔一小时甚至十分钟就要覆盖一轮。这种「小消息、高频次、持续不断」的特征恰好是 Kafka 最擅长的场景。Kafka 在这里不是存数据的是当缓冲和分发枢纽的采集端只负责生产不关心下游是 Flume、实时计算还是可视化大屏。另一个关键点是数据源的选择。常见做法是优先找目标气象站点或公共服务平台的 JSON 接口而不是直接去 parse 网页 HTML。天气类网站基本都带城市代码参数请求 URL 规律、返回结构稳定几分钟就能写完采集逻辑真走到网页解析那一步通常是因为源站的 JSON 接口做了签名校验或风控。我个人做这类项目的第一原则能用 JSON 接口不碰 HTML 解析能抓结构化数据不抓渲染后页面。这直接决定你后面 Kafka 消息体里承载的是干净 JSON 还是需要清洗的半成品。还要考虑采集侧的扩展方式。城市少的时候单机脚本就够了城市增加到几百个后单进程循环的采集周期会拉长到十几分钟这时候要把「按城市取模」的分布式爬虫方式考虑进去——不同机器负责不同城市段各自往同一个 Kafka topic 发送下游 Flume 完全无感。这也是把 Kafka 放在爬虫和存储之间的最大收益采集端可以随时加机器存储端完全不需要跟着变。2.2 requests 采集脚本最小可运行的天气爬虫开发期最省事的组合就是 requests 加 kafka-python一个负责拿数据一个负责发消息。下面这段代码是我常用的最小骨架采集一个城市后立刻发一条消息到 Kafka不落本地文件保持管道尽可能短。import json import time from datetime import datetime import requests from kafka import KafkaProducer # 城市代码 → 城市名实际按你的数据源维护 CITIES { 101010100: 北京, 101020100: 上海, 101040100: 重庆, } def fetch_weather(city_code: str) - dict: # 示例接口地址实际以你对接的源为准 url fhttp://weather.data.example.com/v1/city/{city_code}.html params { appid: your_app_id, version: v1, timestamp: int(time.time()), } resp requests.get(url, paramsparams, timeout10) resp.raise_for_status() return resp.json() def build_record(city_code: str, data: dict) - dict: # 把嵌套 JSON 拉平成一条记录减少下游解析成本 now datetime.now().strftime(%Y-%m-%d %H:%M:%S) return { city_code: city_code, obs_time: now, temperature: data.get(temperature), humidity: data.get(humidity), wind_direction: data.get(wind_direction), raw: json.dumps(data, ensure_asciiFalse), } def send_to_kafka(producer: KafkaProducer, city_code: str, record: dict) - None: future producer.send( topicweather.raw, keycity_code.encode(utf-8), # 同一个城市永远进同一分区 valuejson.dumps(record, ensure_asciiFalse).encode(utf-8), ) # 开发阶段同步等 ack方便第一时间暴露问题 future.get(timeout5) if __name__ __main__: producer KafkaProducer( bootstrap_serverskafka1:9092,kafka2:9092,kafka3:9092, acksall, retries3, linger_ms20, batch_size16384, ) for code, name in CITIES.items(): try: data fetch_weather(code) record build_record(code, data) send_to_kafka(producer, code, record) print(f{name} 采集完成并发送) except Exception as exc: # 采集失败时保留现场不要吞异常 print(f{name} 采集失败: {exc}) continue time.sleep(2) # 限速避免触发反爬这段代码里有几个参数不是随手写的。keycity_code保证北京的数据永远进同一个分区这为后面消费端做城市维度聚合留了后路acksall是采集侧最重要的可靠性开关Leader 副本确认不算数要等 ISR 里所有副本都写入才返回代价是单条消息延迟略高但对天气这种低频场景完全可接受。linger_ms20和batch_size16384配合着看前者是消息在生产者缓冲区最长等 20 毫秒再批量发送后者是攒够 16KB 就提前发。天气数据单条很小如果linger_ms设成 0每条消息都可能触发一次网络往返吞吐上不去设太大又会让消息在内存里干等把「实时」做成「准实时」。20 毫秒是吞吐和延迟之间一个比较中庸的取值流量更低时可以调到 50。2.3 采集到分发producer 端异步入队与流控参数上一节代码里我用的是producer.send()加future.get(timeout5)这其实是一种半异步写法send 本身不阻塞但立刻去等结果。生产环境我一般不这样写而是让 send 之后立刻进入下一个城市的采集把 future 收集起来统一处理。否则每个城市都同步等 ack几百个城市的采集耗时会被网络往返放大好几倍。流控是采集侧最容易忽视的一环。requests 默认没有重试机制遇到 TLS 握手失败或连接池耗尽直接抛异常更隐蔽的是目标源做了限流不报错但返回固定结构的数据比如「请求太频繁」的 JSON。所以采集脚本里必须对响应内容做一层校验而不是只检查 HTTP 状态码。我的习惯是校验返回 JSON 里是否包含预期的字段名比如temperature缺失就重试三次三次都失败就告警并跳过这个城市而不是把脏数据发进 Kafka。另外一个值得一开始就定好的规范消息体里必须带业务时间和采集时间两个字段。业务时间obs_time是天气数据本身代表的时刻采集时间是脚本抓到它的时刻。后面排查延迟时这两个时间戳一减就知道是采集侧慢、Kafka 积压还是 Flume 写入慢。没有这个设计问题发生时你只有一堆来源不明、时间不清的数据排障会非常痛苦。提示采集脚本要记得处理时区。城市接口返回的时间可能是 UTC直接写进 Kafka 会跟本地时间混淆统一转成 UTC 存储展示层再转本地时区是最省事的约定。3. Kafka 实时分发topic 设计、分区策略与消费端顺序性3.1 topic 设计与分区数先定吞吐再定副本消息进了 Kafka第一个要拍板的是 topic 怎么建。天气数据量不大但很多初学者会把 topic 按城市粒度拆——一个城市一个 topic后续要加城市就得不停建 topicFlume 侧也得为每个 topic 各配一个 source维护成本指数级上升。正确做法是只建一个weather.rawtopic所有城市的天气消息混在一起靠消息里的city_code字段区分。下游要拆城市消费端按 key 聚合就行Kafka 分区才是并行度的单位不是 topic。分区数怎么定经验公式是分区数 ≈ 生产端并发度 × 消费端并发度再留一点余量。采集单机跑、Flume 单实例消费的场景6 个分区完全够用如果你计划扩展到分布式爬虫每多一台采集机分区数至少要多出 23 个。分区太多没好处每个分区的 Leader 副本都要占文件句柄和内存分区太少则消费端并发上不去Flume 单线程消费会成为吞吐瓶颈。副本数固定设 2 或 3。天气数据丢了可以重新爬不像交易数据那么致命2 个副本已经能扛单节点宕机如果你们 Kafka 集群本来就是 3 节点起步直接 3 副本坏一台节点不影响生产端 ack。创建 topic 我习惯用命令行动手不用 AdminClient 写代码因为 topic 属于运维配置而不是业务逻辑kafka-topics.sh --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 \ --create \ --topic weather.raw \ --partitions 6 \ --replication-factor 2 \ --config cleanup.policydelete \ --config retention.ms604800000retention.ms604800000表示消息保留 7 天这对天气场景足够了——Flume 是实时消费Kafka 里保留 7 天的意义只是给「消费延迟排障」和「数据重放」留余地。如果你想省磁盘改成 24 小时也可以但重放数据的操作窗口会变小。3.2 生产者配置acks、batch.size、linger.ms 怎么配合上一章的代码里已经给过一组 producer 参数这里把这几个参数之间的配合关系说透因为它们是 Kafka 调优里最容易互相打架的三个参数。acks控制的是「消息写入多少个副本后才算成功」可选值是 0、1、all。采集天气数据建议用 all原因只有一个Flume 消费端是持续运行的如果生产端 ack 机制太松偶发丢失的消息不会报错但 HBase 里就会莫名其妙少几个城市的数据这种问题最难查。用 all 之后任何副本写入失败都会让生产端抛异常你至少能感知到。batch.size和linger.ms是一对配合参数batch 是积攒的字节数上限linger 是积攒的时间上限谁先到就触发发送。天气消息一条算上 JSON 也就 300500 字节batch.size16384意味着要攒 3050 条才批量发一次如果采集频率是每分钟一条这个 batch 永远攒不满。这时候真正起作用的其实是linger_ms——到 20 毫秒就发而不是傻等 batch 攒满。原则是高吞吐场景靠 batch.size 卡边界低吞吐场景靠 linger_ms 卡延迟上限。你的采集频率决定你应该重点调哪个。max.request.size是另一个容易踩的默认值默认 1MB。天气消息远到不了这个值但如果你在build_record里加了raw字段保存原始返回某些接口会把小时预报的 24 条数据一次性返回整个 JSON 可能到几百 KB。我建议关注一下单条消息的实际大小超过 100KB 就要考虑是不是把太多冗余数据塞进了消息体。3.3 消费端多线程与消息顺序性关键词分区是唯一现实解Flume 的 Kafka source 本质上是 Kafka 消费者默认一个 source 实例单线程消费。而 Kafka 的分区是并行度上限一个分区同时只能被一个消费组里的一个消费者线程消费。所以如果你建了 6 个分区但只跑了一个 Flume source剩下的 5 个分区完全空闲吞吐上不去。要让消费端并行起来有两种做法一种是在一台机器上跑多个 Flume agent 实例每个实例的 Kafka source 配同一个group.idKafka 会自动把 6 个分区分配到多个 agent 上另一种是流量没那么大、只是想让单 agent 内多线程处理时用 KafkaConsumer 写自定义消费程序手动把分区分配给多个线程。Kafka 原生不保证消息顺序性因为它只保证「同一分区内有序」。所以要做到天气数据的顺序性——这在高频覆盖场景里其实是伪需求因为一天后最新温度会覆盖旧值——标准做法就是我在 2.2 节里写的生产端按city_code作 key同一个城市永远落同一个分区消费端保证这个分区的消息被同一个线程处理。跨分区无序在天气覆盖场景里没有影响如果换成订单流水这类必须全局有序的业务靠 Kafka 本身做不了得换思路比如并发控制或单分区。提示排查 Kafka 消息积压时优先看消费组的 lag 而不是看磁盘用量。用kafka-consumer-groups.sh --describe --group flume-hbase-group查看每个分区的 current-offset 和 log-end-offsetlag 始终涨说明消费端写入 HBase 有瓶颈lag 不动但数据没入库说明消费端根本没拿到消息问题在别处。4. Flume 收集数据导入 HBasesource、sink 配置与 rowkey 设计4.1 Flume 在管道里的位置为什么不用 Kafka ConnectKafka 和 HBase 之间有很多条路能走常见的有自研消费程序、Apache Flume、Kafka Connect 的 HBase Sink 插件。我做这类采集管道时仍然偏好 Flume原因很落地Flume 天然是「source → channel → sink」三段式故障隔离做得干净。采集源 Kafka 挂了Flume 的 source 会重试但 channel 里的数据不丢HBase 写入超时sink 可以重试而 source 继续消费这种松耦合比一个自研长驻进程更容易运维。Kafka Connect 的问题是 HBase Sink 需要额外下载连接器 jar 且版本匹配经常让人抓狂对「爬虫数据入 HBase」这种轻量场景来说为了一个连接器去维护 Connect 集群不值得。自研消费程序的问题则在于要用 Java 写一个线程安全、支持批量提交、异常恢复的消费者再处理 HBase 连接池工作量至少一天起步Flume 半小时就能跑起来。选 Flume 的前提是你们已经接受了 Java 生态这条路。Flume 和 HBase 都是 JVM 系部署时只要把 HBase 客户端 jar 拷进 Flume 的 lib 目录就能连不需要额外写一行代码做序列化。在小数据量场景下Flume 的内存 channel 足够撑起全链路完全不需要上 Kafka channel 或者 File channel 增加复杂度。4.2 flume.conf 完整配置Kafka source memory channel HBase sink下面是完整可用的 Flume 配置文件。agent 名我用weather-agent三段配置分别是 Kafka 消费者、内存 channel、HBase sink。# flume-weather.conf # agent 名称weather-agent weather-agent.sources kafka-source weather-agent.channels hbase-channel weather-agent.sinks hbase-sink # ---------------- sourceKafka 消费者 ---------------- weather-agent.sources.kafka-source.type org.apache.flume.source.kafka.KafkaSource weather-agent.sources.kafka-source.kafka.bootstrap.servers kafka1:9092,kafka2:9092,kafka3:9092 weather-agent.sources.kafka-source.kafka.topic weather.raw weather-agent.sources.kafka-source.kafka.consumer.group.id flume-hbase-group weather-agent.sources.kafka-source.kafka.consumer.auto.offset.reset earliest weather-agent.sources.kafka-source.batchSize 500 weather-agent.sources.kafka-source.batchDurationMillis 2000 # ---------------- channel内存 channel ---------------- weather-agent.channels.hbase-channel.type memory weather-agent.channels.hbase-channel.capacity 20000 weather-agent.channels.hbase-channel.transactionCapacity 1000 weather-agent.channels.hbase-channel.byteCapacity 52428800 # ---------------- sinkHBase 写入 ---------------- weather-agent.sinks.hbase-sink.type org.apache.flume.sink.hbase.HBaseSink weather-agent.sinks.hbase-sink.table weather_hbase weather-agent.sinks.hbase-sink.columnFamily cf weather-agent.sinks.hbase-sink.serializer org.apache.flume.sink.hbase.RegexHbaseEventSerializer weather-agent.sinks.hbase-sink.serializer.regex city_code:(([^])).*temperature:(([^])) weather-agent.sinks.hbase-sink.serializer.columnNames city_code,temperature weather-agent.sinks.hbase-sink.zkQuorum hbase-zk1:2181,hbase-zk2:2181,hbase-zk3:2181 weather-agent.sinks.hbase-sink.batchSize 1000逐段说下配置意图。source 里的kafka.consumer.auto.offset.resetearliest表示消费组没有提交过 offset 时从最早消息开始消费这是 Flume 首次启动时「把 Kafka 里积压的天气数据补进 HBase」的关键如果设成latest新部署时老消息会被跳过数据静默丢失。channel 的三个参数要一起看capacity20000是 channel 能缓存的最大事件条数transactionCapacity1000是 source 一次事务能写入 channel 的最大条数byteCapacity是总字节上限。memory channel 看起来简单但有个隐蔽风险——agent 进程崩溃时 channel 里未写入 HBase 的数据直接丢失这是整套管道里唯一的非持久化环节。Flume 官方给的取舍是memory channel 换吞吐file channel 换可靠性。天气数据丢了能重爬所以我选吞吐如果你接的是订单、日志这类不能丢的来源老老实实用 file channel。sink 里的RegexHbaseEventSerializer是 Flume 自带的序列化器用正则从消息体里提取字段把匹配到的值写进 HBase 列。这个正则要求消息体里的 JSON 是单行的如果build_record里json.dumps没有压缩而是用了indent2正则就匹配不上sink 会把整条消息原样塞进 HBase 的一个列里——数据不丢但是结构完全是乱的。Flume 的 serializer 正则与采集端的 JSON 格式是一对强耦合改任何一边都得同步检查另一边。4.3 HBase 表设计与 rowkey 规则预分区和 WAL 路径要提前想好HBase 表设计是这套管道里最需要提前规划的一步因为建表后改 rowkey 规则代价极大。天气场景的查询模式通常是「查某城市某天的温度曲线」rowkey 的经典设计是把城市 code 放前面、时间戳放后面但因为 HBase 按 rowkey 字典序分布直接存原始时间戳会导致所有写入都落在最后一个 Region 上形成热点。应对热点的标准做法是给 rowkey 加哈希前缀。我一般这样设计rowkey 城市 code 的哈希值取模 10 作为前缀 城市 code 本身 倒序时间戳。前缀把数据均匀打散到 10 个 Region倒序时间戳让同一城市的最新数据排在前面扫表时能快速取到最新记录。建表时用预分区建 10 个 Region而不是等数据量大了再手动拆分。hbase shell EOF create weather_hbase, {NAME cf, VERSIONS 3, COMPRESSION GZ}, {SPLITS [0, 1, 2, 3, 4, 5, 6, 7, 8]} EOFVERSIONS 3要解释一下HBase 默认只保留一个版本的单元格数据如果你想让管道重放或者排查历史写错数据保留 3 个版本能直接查到覆盖前的旧值这对「爬虫数据反复重试采集」的场景很实用因为同一时间点可能跑到多次覆盖写入。SPLITS列表里有 9 个切分点会把表预分成 10 个 Region前缀 09 正好各落一个 Region。WAL 是另一个要在建表前确认的点确认 HBase 的hbase.wal.dir路径配置无误这个在避坑章节展开讲。5. 这套管道的高频疑难与避坑排查现象、原因、解决办法5.1 采集脚本不报错Kafka 里全是空壳数据现象脚本日志每行都显示「采集完成并发送」Kafka topic 里消息数也在涨但 Flume 写进 HBase 后查不到任何业务字段temperature和humidity全是 null。原因天气接口返回字段名跟代码里用的 key 对不上。真实接口的嵌套结构里字段名可能是temp、hum而build_record里写的是temperature、humiditydata.get(temperature)返回 None异常被try except吞了或根本没校验空壳消息照样进了 Kafka。这锅不在管道在采集侧没有做字段完整性校验。解决在send_to_kafka之前加一层校验——if record[temperature] is None: raise ValueError。同时把raw字段保留在消息体里出问题时直接用 hbase shell get 单条数据看原始 JSON字段名对不对一目了然。5.2 HBase 写入全部压在最后一个 Region热点写把集群打慢现象Flume 的 sink 写入耗时越来越长HBase 监控里某个 RegionServer 的写请求 QPS 是其他节点的十几倍compaction 频繁触发。原因建表时没做预分区或者做了预分区但 rowkey 没加哈希前缀。rowkey 里时间戳是递增的所有新写入天然落在最后一个 Region数据量一大必然形成写热点。Flume 侧看起来是「写入慢」本质是 HBase 侧的负载不均衡。解决回 4.3 节的方案——预分区 前缀取模。如果你已经建表且不想重建反向补就的姿势是停 Flume 写入对现有表做split找出热点 Region再用major_compact合并小文件但这是打补丁正确解法就是删表重建并让 Flume 从 Kafka 里重放前提是 retention 还在保留期内所以前面才建议保留 7 天。5.3 Kafka 消息延迟高企linger.ms 设太大实时性被自己破坏现象Kafka 生产端从请求到 ack 耗时正常但消费者看到的每条消息都比业务时间晚 15 秒吞吐不高但延迟很大。原因producer 的linger_ms设得过大比如小白抄配置抄成了linger.ms500。天气数据频率低batch 永远攒不满每条消息都在缓冲区干等 500 毫秒才发。这不是管道慢是生产端「为了批量牺牲了实时」而天气这个场景根本不需要那么大的批量。解决把linger_ms调回 2050 毫秒。同时可以看消费端fetch.max.wait.msKafka 消费者默认是 500 毫秒才拉一次如果数据真要求秒级到达把这个参数同步调到 100 毫秒以内。这类问题排查时用「业务时间和采集时间之差」最直接差值稳定且偏大优先怀疑生产端或消费端的等待参数。5.4 HBase WAL 预写日志异常导致 Flume sink 反复失败现象Flume 日志里刷org.apache.hadoop.hbase.wal.WAL相关异常sink 一直重试写入同一个 batchHBase 侧对应 RegionServer 的日志里有 WAL 同步失败记录。原因HBase 的 WAL 路径配置在hbase-site.xml的hbase.wal.dir实际使用时通常是 HDFS 上的一个目录。路径对应的 HDFS 目录没有写权限或者数据节点磁盘空间不足WAL 写不进去RegionServer 拒绝新写入。Flume 的 HBaseSink 重试机制在这种场景下是无效的——问题不在 Flume在 HBase 自身的存储层。解决先确认 HDFS 目录存在且属主正确通常用hdfs dfs -chown hbase:hbase /apps/hbase/wal修正属主再查数据节点的磁盘剩余空间低于阈值时 HBase 会主动进入只读保护模式。修复后不要重启整个 HBase 集群只重启出问题的 RegionServer。Flume 侧把hbase-sink.batchSize调小到 500减少单次事务跨度WAL 再次抖动时重试成本更低。5.5 反爬拦截采集侧 IP 被封管道再稳也是空转现象Kafka 里消息数骤降甚至为 0Flume 完全空闲HBase 最近一小时没有新数据。采集脚本没有报错但请求目标是网站的反爬页。原因天气数据源的频率限制通常是温和的但如果采集端没有限速或者单 IP 并发太快源站会返回验证页面或者直接把请求重定向到静态页。requests 不会告诉你被反爬了它只按 HTTP 200 处理。解决采集侧限速是基础2.2 节代码里的time.sleep(2)就是干这个的。再加一层保险检查响应内容里是否有预期的业务字段没有就视为反爬拦截并指数退避连退三次就告警。分布式爬虫场景里多台机器共用一个出口 IP 是最常见的封 IP 原因把出口 IP 分散到不同城市节点比挂代理池稳得多也省得多。管道本身在这种场景下没有可调的地方问题永远在采集侧。6. 进阶验证从端到端延迟测量到重放机制6.1 端到端延迟验证从 producer 时间戳到 HBase cell管道搭完不能只看「数据进 HBase 了」要量化端到端延迟。我在每条消息里已经写了obs_time和采集时间HBase 侧在建表时设了VERSIONS 3这里再利用 HBase 的 cell 时间戳做一次交叉验证# 取某个城市最新一条记录 hbase shell EOF get weather_hbase, rowkey, {COLUMN cf:obs_time, TIMESTAMP 1700000000000} EOF把返回的Timestamp与 Kafka 生产端的发送时间做差就是这条消息从「脚步离开采集机」到「落进 HBase」的全链路耗时。通常在几百毫秒到几秒之间如果稳定超过 5 秒按 5.3 节的方法逐段检查生产端、Kafka 拉取参数和 Flume 的 batchDurationMillis。6.2 数据完整性抽查与重放给管道留一剂后悔药数据管道最终交付的不只是代码还有「数据出事了怎么补」的能力。我每做一个采集管道都会把重放流程写进文档Kafka 消息保留期内Flume 消费组重置 offset就能把指定时间段的数据重新导一遍不需要重新爬源站。# 重置消费组到 30 分钟前实现精准重放 kafka-consumer-groups.sh --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 \ --group flume-hbase-group \ --topic weather.raw \ --reset-offsets --to-datetime 2025-01-01T10:00:00.000 \ --execute这个操作是我踩过坑之后养成的习惯。一次误删 HBase 表后我靠这条命令从 Kafka 里把消息全部重放回来分钟级恢复了数据。先删表会连 WAL 一起丢但 Kafka 里的消息还热着这就是当初坚持「Kafka 留 7 天」的回报。从那以后任何采集管道上线我都会在运维手册里写清「误删后的重放流程」而不是等事故来了临场补。希望帮到你。本文还有配套的精品资源点击获取 从爬虫到HBase：Kafka与Flume构建实时天气数据管道
简介面向天气数据实时采集与处理场景这份项目包以完整 Java 工程形式演示了从爬虫抓取天气数据经 Kafka 实时分发再由 Flume 消费并导入 HBase随后通过 Hive 建立映射、Superset 展示分析结果的完整链路。适合大数据实时数仓、消息中间件与 NoSQL 存储方向的开发者参考。压缩包共 17 个文件含 11 个 Java 源码、1 个 XML 工程配置、3 张 PNG 架构示意图及说明文档整体约 985KB体量轻、目录清晰便于快速阅读工程结构。已有 152 人学习下载。该示例覆盖数据源接入、Topic 设计、Flume 导入、HBase 表结构及 Hive 外表映射等关键点可作为实际项目的可运行参考并二次扩展。1. 天气爬虫采集到 HBase一条实时数据管道的完整落地路径「天气爬虫采集Kafka 实时分发Flume 收集数据导入 HBase」——这串标题看着像三个工具拼在一起实际是一条完整的数据管道采集端用爬虫把天气数据拿下来经 Kafka 做实时缓冲与分发再由 Flume 落进 HBase 供查询和分析。这套组合适合两类人一类在做城市天气、气象实况类的数据服务另一类想照葫芦画瓢搭「爬虫 → 消息队列 → 列式存储」这套通用实时管道。本文按我实际落地的顺序讲采集代码怎么写、Kafka topic 和分区怎么设、Flume 配置文件怎么填以及最让我翻车的那几个坑。2. 天气爬虫采集先想清楚「采什么、发去哪」再写第一行代码2.1 数据源与采集策略为什么天气数据天然适合这条管道天气数据和爬虫常见的电商、社交数据不一样它有两个对管道设计影响很大的特点单条记录小、字段结构固定但更新频率高——温度、湿度、风力每隔一小时甚至十分钟就要覆盖一轮。这种「小消息、高频次、持续不断」的特征恰好是 Kafka 最擅长的场景。Kafka 在这里不是存数据的是当缓冲和分发枢纽的采集端只负责生产不关心下游是 Flume、实时计算还是可视化大屏。另一个关键点是数据源的选择。常见做法是优先找目标气象站点或公共服务平台的 JSON 接口而不是直接去 parse 网页 HTML。天气类网站基本都带城市代码参数请求 URL 规律、返回结构稳定几分钟就能写完采集逻辑真走到网页解析那一步通常是因为源站的 JSON 接口做了签名校验或风控。我个人做这类项目的第一原则能用 JSON 接口不碰 HTML 解析能抓结构化数据不抓渲染后页面。这直接决定你后面 Kafka 消息体里承载的是干净 JSON 还是需要清洗的半成品。还要考虑采集侧的扩展方式。城市少的时候单机脚本就够了城市增加到几百个后单进程循环的采集周期会拉长到十几分钟这时候要把「按城市取模」的分布式爬虫方式考虑进去——不同机器负责不同城市段各自往同一个 Kafka topic 发送下游 Flume 完全无感。这也是把 Kafka 放在爬虫和存储之间的最大收益采集端可以随时加机器存储端完全不需要跟着变。2.2 requests 采集脚本最小可运行的天气爬虫开发期最省事的组合就是 requests 加 kafka-python一个负责拿数据一个负责发消息。下面这段代码是我常用的最小骨架采集一个城市后立刻发一条消息到 Kafka不落本地文件保持管道尽可能短。import json import time from datetime import datetime import requests from kafka import KafkaProducer # 城市代码 → 城市名实际按你的数据源维护 CITIES { 101010100: 北京, 101020100: 上海, 101040100: 重庆, } def fetch_weather(city_code: str) - dict: # 示例接口地址实际以你对接的源为准 url fhttp://weather.data.example.com/v1/city/{city_code}.html params { appid: your_app_id, version: v1, timestamp: int(time.time()), } resp requests.get(url, paramsparams, timeout10) resp.raise_for_status() return resp.json() def build_record(city_code: str, data: dict) - dict: # 把嵌套 JSON 拉平成一条记录减少下游解析成本 now datetime.now().strftime(%Y-%m-%d %H:%M:%S) return { city_code: city_code, obs_time: now, temperature: data.get(temperature), humidity: data.get(humidity), wind_direction: data.get(wind_direction), raw: json.dumps(data, ensure_asciiFalse), } def send_to_kafka(producer: KafkaProducer, city_code: str, record: dict) - None: future producer.send( topicweather.raw, keycity_code.encode(utf-8), # 同一个城市永远进同一分区 valuejson.dumps(record, ensure_asciiFalse).encode(utf-8), ) # 开发阶段同步等 ack方便第一时间暴露问题 future.get(timeout5) if __name__ __main__: producer KafkaProducer( bootstrap_serverskafka1:9092,kafka2:9092,kafka3:9092, acksall, retries3, linger_ms20, batch_size16384, ) for code, name in CITIES.items(): try: data fetch_weather(code) record build_record(code, data) send_to_kafka(producer, code, record) print(f{name} 采集完成并发送) except Exception as exc: # 采集失败时保留现场不要吞异常 print(f{name} 采集失败: {exc}) continue time.sleep(2) # 限速避免触发反爬这段代码里有几个参数不是随手写的。keycity_code保证北京的数据永远进同一个分区这为后面消费端做城市维度聚合留了后路acksall是采集侧最重要的可靠性开关Leader 副本确认不算数要等 ISR 里所有副本都写入才返回代价是单条消息延迟略高但对天气这种低频场景完全可接受。linger_ms20和batch_size16384配合着看前者是消息在生产者缓冲区最长等 20 毫秒再批量发送后者是攒够 16KB 就提前发。天气数据单条很小如果linger_ms设成 0每条消息都可能触发一次网络往返吞吐上不去设太大又会让消息在内存里干等把「实时」做成「准实时」。20 毫秒是吞吐和延迟之间一个比较中庸的取值流量更低时可以调到 50。2.3 采集到分发producer 端异步入队与流控参数上一节代码里我用的是producer.send()加future.get(timeout5)这其实是一种半异步写法send 本身不阻塞但立刻去等结果。生产环境我一般不这样写而是让 send 之后立刻进入下一个城市的采集把 future 收集起来统一处理。否则每个城市都同步等 ack几百个城市的采集耗时会被网络往返放大好几倍。流控是采集侧最容易忽视的一环。requests 默认没有重试机制遇到 TLS 握手失败或连接池耗尽直接抛异常更隐蔽的是目标源做了限流不报错但返回固定结构的数据比如「请求太频繁」的 JSON。所以采集脚本里必须对响应内容做一层校验而不是只检查 HTTP 状态码。我的习惯是校验返回 JSON 里是否包含预期的字段名比如temperature缺失就重试三次三次都失败就告警并跳过这个城市而不是把脏数据发进 Kafka。另外一个值得一开始就定好的规范消息体里必须带业务时间和采集时间两个字段。业务时间obs_time是天气数据本身代表的时刻采集时间是脚本抓到它的时刻。后面排查延迟时这两个时间戳一减就知道是采集侧慢、Kafka 积压还是 Flume 写入慢。没有这个设计问题发生时你只有一堆来源不明、时间不清的数据排障会非常痛苦。提示采集脚本要记得处理时区。城市接口返回的时间可能是 UTC直接写进 Kafka 会跟本地时间混淆统一转成 UTC 存储展示层再转本地时区是最省事的约定。3. Kafka 实时分发topic 设计、分区策略与消费端顺序性3.1 topic 设计与分区数先定吞吐再定副本消息进了 Kafka第一个要拍板的是 topic 怎么建。天气数据量不大但很多初学者会把 topic 按城市粒度拆——一个城市一个 topic后续要加城市就得不停建 topicFlume 侧也得为每个 topic 各配一个 source维护成本指数级上升。正确做法是只建一个weather.rawtopic所有城市的天气消息混在一起靠消息里的city_code字段区分。下游要拆城市消费端按 key 聚合就行Kafka 分区才是并行度的单位不是 topic。分区数怎么定经验公式是分区数 ≈ 生产端并发度 × 消费端并发度再留一点余量。采集单机跑、Flume 单实例消费的场景6 个分区完全够用如果你计划扩展到分布式爬虫每多一台采集机分区数至少要多出 23 个。分区太多没好处每个分区的 Leader 副本都要占文件句柄和内存分区太少则消费端并发上不去Flume 单线程消费会成为吞吐瓶颈。副本数固定设 2 或 3。天气数据丢了可以重新爬不像交易数据那么致命2 个副本已经能扛单节点宕机如果你们 Kafka 集群本来就是 3 节点起步直接 3 副本坏一台节点不影响生产端 ack。创建 topic 我习惯用命令行动手不用 AdminClient 写代码因为 topic 属于运维配置而不是业务逻辑kafka-topics.sh --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 \ --create \ --topic weather.raw \ --partitions 6 \ --replication-factor 2 \ --config cleanup.policydelete \ --config retention.ms604800000retention.ms604800000表示消息保留 7 天这对天气场景足够了——Flume 是实时消费Kafka 里保留 7 天的意义只是给「消费延迟排障」和「数据重放」留余地。如果你想省磁盘改成 24 小时也可以但重放数据的操作窗口会变小。3.2 生产者配置acks、batch.size、linger.ms 怎么配合上一章的代码里已经给过一组 producer 参数这里把这几个参数之间的配合关系说透因为它们是 Kafka 调优里最容易互相打架的三个参数。acks控制的是「消息写入多少个副本后才算成功」可选值是 0、1、all。采集天气数据建议用 all原因只有一个Flume 消费端是持续运行的如果生产端 ack 机制太松偶发丢失的消息不会报错但 HBase 里就会莫名其妙少几个城市的数据这种问题最难查。用 all 之后任何副本写入失败都会让生产端抛异常你至少能感知到。batch.size和linger.ms是一对配合参数batch 是积攒的字节数上限linger 是积攒的时间上限谁先到就触发发送。天气消息一条算上 JSON 也就 300500 字节batch.size16384意味着要攒 3050 条才批量发一次如果采集频率是每分钟一条这个 batch 永远攒不满。这时候真正起作用的其实是linger_ms——到 20 毫秒就发而不是傻等 batch 攒满。原则是高吞吐场景靠 batch.size 卡边界低吞吐场景靠 linger_ms 卡延迟上限。你的采集频率决定你应该重点调哪个。max.request.size是另一个容易踩的默认值默认 1MB。天气消息远到不了这个值但如果你在build_record里加了raw字段保存原始返回某些接口会把小时预报的 24 条数据一次性返回整个 JSON 可能到几百 KB。我建议关注一下单条消息的实际大小超过 100KB 就要考虑是不是把太多冗余数据塞进了消息体。3.3 消费端多线程与消息顺序性关键词分区是唯一现实解Flume 的 Kafka source 本质上是 Kafka 消费者默认一个 source 实例单线程消费。而 Kafka 的分区是并行度上限一个分区同时只能被一个消费组里的一个消费者线程消费。所以如果你建了 6 个分区但只跑了一个 Flume source剩下的 5 个分区完全空闲吞吐上不去。要让消费端并行起来有两种做法一种是在一台机器上跑多个 Flume agent 实例每个实例的 Kafka source 配同一个group.idKafka 会自动把 6 个分区分配到多个 agent 上另一种是流量没那么大、只是想让单 agent 内多线程处理时用 KafkaConsumer 写自定义消费程序手动把分区分配给多个线程。Kafka 原生不保证消息顺序性因为它只保证「同一分区内有序」。所以要做到天气数据的顺序性——这在高频覆盖场景里其实是伪需求因为一天后最新温度会覆盖旧值——标准做法就是我在 2.2 节里写的生产端按city_code作 key同一个城市永远落同一个分区消费端保证这个分区的消息被同一个线程处理。跨分区无序在天气覆盖场景里没有影响如果换成订单流水这类必须全局有序的业务靠 Kafka 本身做不了得换思路比如并发控制或单分区。提示排查 Kafka 消息积压时优先看消费组的 lag 而不是看磁盘用量。用kafka-consumer-groups.sh --describe --group flume-hbase-group查看每个分区的 current-offset 和 log-end-offsetlag 始终涨说明消费端写入 HBase 有瓶颈lag 不动但数据没入库说明消费端根本没拿到消息问题在别处。4. Flume 收集数据导入 HBasesource、sink 配置与 rowkey 设计4.1 Flume 在管道里的位置为什么不用 Kafka ConnectKafka 和 HBase 之间有很多条路能走常见的有自研消费程序、Apache Flume、Kafka Connect 的 HBase Sink 插件。我做这类采集管道时仍然偏好 Flume原因很落地Flume 天然是「source → channel → sink」三段式故障隔离做得干净。采集源 Kafka 挂了Flume 的 source 会重试但 channel 里的数据不丢HBase 写入超时sink 可以重试而 source 继续消费这种松耦合比一个自研长驻进程更容易运维。Kafka Connect 的问题是 HBase Sink 需要额外下载连接器 jar 且版本匹配经常让人抓狂对「爬虫数据入 HBase」这种轻量场景来说为了一个连接器去维护 Connect 集群不值得。自研消费程序的问题则在于要用 Java 写一个线程安全、支持批量提交、异常恢复的消费者再处理 HBase 连接池工作量至少一天起步Flume 半小时就能跑起来。选 Flume 的前提是你们已经接受了 Java 生态这条路。Flume 和 HBase 都是 JVM 系部署时只要把 HBase 客户端 jar 拷进 Flume 的 lib 目录就能连不需要额外写一行代码做序列化。在小数据量场景下Flume 的内存 channel 足够撑起全链路完全不需要上 Kafka channel 或者 File channel 增加复杂度。4.2 flume.conf 完整配置Kafka source memory channel HBase sink下面是完整可用的 Flume 配置文件。agent 名我用weather-agent三段配置分别是 Kafka 消费者、内存 channel、HBase sink。# flume-weather.conf # agent 名称weather-agent weather-agent.sources kafka-source weather-agent.channels hbase-channel weather-agent.sinks hbase-sink # ---------------- sourceKafka 消费者 ---------------- weather-agent.sources.kafka-source.type org.apache.flume.source.kafka.KafkaSource weather-agent.sources.kafka-source.kafka.bootstrap.servers kafka1:9092,kafka2:9092,kafka3:9092 weather-agent.sources.kafka-source.kafka.topic weather.raw weather-agent.sources.kafka-source.kafka.consumer.group.id flume-hbase-group weather-agent.sources.kafka-source.kafka.consumer.auto.offset.reset earliest weather-agent.sources.kafka-source.batchSize 500 weather-agent.sources.kafka-source.batchDurationMillis 2000 # ---------------- channel内存 channel ---------------- weather-agent.channels.hbase-channel.type memory weather-agent.channels.hbase-channel.capacity 20000 weather-agent.channels.hbase-channel.transactionCapacity 1000 weather-agent.channels.hbase-channel.byteCapacity 52428800 # ---------------- sinkHBase 写入 ---------------- weather-agent.sinks.hbase-sink.type org.apache.flume.sink.hbase.HBaseSink weather-agent.sinks.hbase-sink.table weather_hbase weather-agent.sinks.hbase-sink.columnFamily cf weather-agent.sinks.hbase-sink.serializer org.apache.flume.sink.hbase.RegexHbaseEventSerializer weather-agent.sinks.hbase-sink.serializer.regex city_code:(([^])).*temperature:(([^])) weather-agent.sinks.hbase-sink.serializer.columnNames city_code,temperature weather-agent.sinks.hbase-sink.zkQuorum hbase-zk1:2181,hbase-zk2:2181,hbase-zk3:2181 weather-agent.sinks.hbase-sink.batchSize 1000逐段说下配置意图。source 里的kafka.consumer.auto.offset.resetearliest表示消费组没有提交过 offset 时从最早消息开始消费这是 Flume 首次启动时「把 Kafka 里积压的天气数据补进 HBase」的关键如果设成latest新部署时老消息会被跳过数据静默丢失。channel 的三个参数要一起看capacity20000是 channel 能缓存的最大事件条数transactionCapacity1000是 source 一次事务能写入 channel 的最大条数byteCapacity是总字节上限。memory channel 看起来简单但有个隐蔽风险——agent 进程崩溃时 channel 里未写入 HBase 的数据直接丢失这是整套管道里唯一的非持久化环节。Flume 官方给的取舍是memory channel 换吞吐file channel 换可靠性。天气数据丢了能重爬所以我选吞吐如果你接的是订单、日志这类不能丢的来源老老实实用 file channel。sink 里的RegexHbaseEventSerializer是 Flume 自带的序列化器用正则从消息体里提取字段把匹配到的值写进 HBase 列。这个正则要求消息体里的 JSON 是单行的如果build_record里json.dumps没有压缩而是用了indent2正则就匹配不上sink 会把整条消息原样塞进 HBase 的一个列里——数据不丢但是结构完全是乱的。Flume 的 serializer 正则与采集端的 JSON 格式是一对强耦合改任何一边都得同步检查另一边。4.3 HBase 表设计与 rowkey 规则预分区和 WAL 路径要提前想好HBase 表设计是这套管道里最需要提前规划的一步因为建表后改 rowkey 规则代价极大。天气场景的查询模式通常是「查某城市某天的温度曲线」rowkey 的经典设计是把城市 code 放前面、时间戳放后面但因为 HBase 按 rowkey 字典序分布直接存原始时间戳会导致所有写入都落在最后一个 Region 上形成热点。应对热点的标准做法是给 rowkey 加哈希前缀。我一般这样设计rowkey 城市 code 的哈希值取模 10 作为前缀 城市 code 本身 倒序时间戳。前缀把数据均匀打散到 10 个 Region倒序时间戳让同一城市的最新数据排在前面扫表时能快速取到最新记录。建表时用预分区建 10 个 Region而不是等数据量大了再手动拆分。hbase shell EOF create weather_hbase, {NAME cf, VERSIONS 3, COMPRESSION GZ}, {SPLITS [0, 1, 2, 3, 4, 5, 6, 7, 8]} EOFVERSIONS 3要解释一下HBase 默认只保留一个版本的单元格数据如果你想让管道重放或者排查历史写错数据保留 3 个版本能直接查到覆盖前的旧值这对「爬虫数据反复重试采集」的场景很实用因为同一时间点可能跑到多次覆盖写入。SPLITS列表里有 9 个切分点会把表预分成 10 个 Region前缀 09 正好各落一个 Region。WAL 是另一个要在建表前确认的点确认 HBase 的hbase.wal.dir路径配置无误这个在避坑章节展开讲。5. 这套管道的高频疑难与避坑排查现象、原因、解决办法5.1 采集脚本不报错Kafka 里全是空壳数据现象脚本日志每行都显示「采集完成并发送」Kafka topic 里消息数也在涨但 Flume 写进 HBase 后查不到任何业务字段temperature和humidity全是 null。原因天气接口返回字段名跟代码里用的 key 对不上。真实接口的嵌套结构里字段名可能是temp、hum而build_record里写的是temperature、humiditydata.get(temperature)返回 None异常被try except吞了或根本没校验空壳消息照样进了 Kafka。这锅不在管道在采集侧没有做字段完整性校验。解决在send_to_kafka之前加一层校验——if record[temperature] is None: raise ValueError。同时把raw字段保留在消息体里出问题时直接用 hbase shell get 单条数据看原始 JSON字段名对不对一目了然。5.2 HBase 写入全部压在最后一个 Region热点写把集群打慢现象Flume 的 sink 写入耗时越来越长HBase 监控里某个 RegionServer 的写请求 QPS 是其他节点的十几倍compaction 频繁触发。原因建表时没做预分区或者做了预分区但 rowkey 没加哈希前缀。rowkey 里时间戳是递增的所有新写入天然落在最后一个 Region数据量一大必然形成写热点。Flume 侧看起来是「写入慢」本质是 HBase 侧的负载不均衡。解决回 4.3 节的方案——预分区 前缀取模。如果你已经建表且不想重建反向补就的姿势是停 Flume 写入对现有表做split找出热点 Region再用major_compact合并小文件但这是打补丁正确解法就是删表重建并让 Flume 从 Kafka 里重放前提是 retention 还在保留期内所以前面才建议保留 7 天。5.3 Kafka 消息延迟高企linger.ms 设太大实时性被自己破坏现象Kafka 生产端从请求到 ack 耗时正常但消费者看到的每条消息都比业务时间晚 15 秒吞吐不高但延迟很大。原因producer 的linger_ms设得过大比如小白抄配置抄成了linger.ms500。天气数据频率低batch 永远攒不满每条消息都在缓冲区干等 500 毫秒才发。这不是管道慢是生产端「为了批量牺牲了实时」而天气这个场景根本不需要那么大的批量。解决把linger_ms调回 2050 毫秒。同时可以看消费端fetch.max.wait.msKafka 消费者默认是 500 毫秒才拉一次如果数据真要求秒级到达把这个参数同步调到 100 毫秒以内。这类问题排查时用「业务时间和采集时间之差」最直接差值稳定且偏大优先怀疑生产端或消费端的等待参数。5.4 HBase WAL 预写日志异常导致 Flume sink 反复失败现象Flume 日志里刷org.apache.hadoop.hbase.wal.WAL相关异常sink 一直重试写入同一个 batchHBase 侧对应 RegionServer 的日志里有 WAL 同步失败记录。原因HBase 的 WAL 路径配置在hbase-site.xml的hbase.wal.dir实际使用时通常是 HDFS 上的一个目录。路径对应的 HDFS 目录没有写权限或者数据节点磁盘空间不足WAL 写不进去RegionServer 拒绝新写入。Flume 的 HBaseSink 重试机制在这种场景下是无效的——问题不在 Flume在 HBase 自身的存储层。解决先确认 HDFS 目录存在且属主正确通常用hdfs dfs -chown hbase:hbase /apps/hbase/wal修正属主再查数据节点的磁盘剩余空间低于阈值时 HBase 会主动进入只读保护模式。修复后不要重启整个 HBase 集群只重启出问题的 RegionServer。Flume 侧把hbase-sink.batchSize调小到 500减少单次事务跨度WAL 再次抖动时重试成本更低。5.5 反爬拦截采集侧 IP 被封管道再稳也是空转现象Kafka 里消息数骤降甚至为 0Flume 完全空闲HBase 最近一小时没有新数据。采集脚本没有报错但请求目标是网站的反爬页。原因天气数据源的频率限制通常是温和的但如果采集端没有限速或者单 IP 并发太快源站会返回验证页面或者直接把请求重定向到静态页。requests 不会告诉你被反爬了它只按 HTTP 200 处理。解决采集侧限速是基础2.2 节代码里的time.sleep(2)就是干这个的。再加一层保险检查响应内容里是否有预期的业务字段没有就视为反爬拦截并指数退避连退三次就告警。分布式爬虫场景里多台机器共用一个出口 IP 是最常见的封 IP 原因把出口 IP 分散到不同城市节点比挂代理池稳得多也省得多。管道本身在这种场景下没有可调的地方问题永远在采集侧。6. 进阶验证从端到端延迟测量到重放机制6.1 端到端延迟验证从 producer 时间戳到 HBase cell管道搭完不能只看「数据进 HBase 了」要量化端到端延迟。我在每条消息里已经写了obs_time和采集时间HBase 侧在建表时设了VERSIONS 3这里再利用 HBase 的 cell 时间戳做一次交叉验证# 取某个城市最新一条记录 hbase shell EOF get weather_hbase, rowkey, {COLUMN cf:obs_time, TIMESTAMP 1700000000000} EOF把返回的Timestamp与 Kafka 生产端的发送时间做差就是这条消息从「脚步离开采集机」到「落进 HBase」的全链路耗时。通常在几百毫秒到几秒之间如果稳定超过 5 秒按 5.3 节的方法逐段检查生产端、Kafka 拉取参数和 Flume 的 batchDurationMillis。6.2 数据完整性抽查与重放给管道留一剂后悔药数据管道最终交付的不只是代码还有「数据出事了怎么补」的能力。我每做一个采集管道都会把重放流程写进文档Kafka 消息保留期内Flume 消费组重置 offset就能把指定时间段的数据重新导一遍不需要重新爬源站。# 重置消费组到 30 分钟前实现精准重放 kafka-consumer-groups.sh --bootstrap-server kafka1:9092,kafka2:9092,kafka3:9092 \ --group flume-hbase-group \ --topic weather.raw \ --reset-offsets --to-datetime 2025-01-01T10:00:00.000 \ --execute这个操作是我踩过坑之后养成的习惯。一次误删 HBase 表后我靠这条命令从 Kafka 里把消息全部重放回来分钟级恢复了数据。先删表会连 WAL 一起丢但 Kafka 里的消息还热着这就是当初坚持「Kafka 留 7 天」的回报。从那以后任何采集管道上线我都会在运维手册里写清「误删后的重放流程」而不是等事故来了临场补。希望帮到你。本文还有配套的精品资源点击获取