做物联网设备监控那会儿团队最头疼的就是链路越拉越长采集端每秒几十万条传感器指标进到 Kafka在线计算跟不上存储扛不住查询那边还经常超时。前前后后对比了不少方案最后定型下来的组合就是 Flink 负责实时计算Lindorm TSDB 负责时序数据落地和查询。这套链路从零搭起来以后跑了大半年这里把我的思路、代码和踩过的坑完整复盘一遍。文章会覆盖架构选型、表模型设计、自定义 Sink 的细节以及上线后的排查经验适合正在做时序数据处理、运维监控、IoT 数据平台的开发同学参考不论你是刚接触 Flink 还是已经有基础里面有不少东西可以直接抄作业。1. 架构设计思路为什么把 Flink 和 Lindorm TSDB 放在一起1.1 时序数据管线的三个老问题时序数据和普通业务数据不太一样。首先是数据增长几乎不带停的传感器每秒钟产生一条数据一个月就是上亿条关系型数据库别说查询了写入都扛不住。其次是数据形态有强烈的时空属性几乎所有查询都是“某个设备、某个时间段内”的聚合分析天然需要按时间维度组织和检索。最后是写多读少但读的实时性要求高比如大屏上的温度曲线、告警系统里的最近五分钟均值这些都需要秒级返回。我在做这套方案之前试过先用 Kafka 把原始数据存下来再由业务侧程序直接消费 Kafka 算指标。这样做的缺点很明显各种聚合逻辑要自己写状态管理数据重复消费时去重也麻烦而且下游查询只能扫 Kafka时间范围一大就非常慢。后来意识到必须拆分职责实时计算和时序存储各干各的事谁擅长什么就干什么。1.2 Lindorm TSDB 在存储侧的价值选 Lindorm TSDB 之前我也对比过 OpenTSDB、InfluxDB 自建集群和 HBase 自研方案。自建 InfluxDB 在集群规模和压缩成本上会踩不少坑HBase 则需要投入大量人力去设计 rowkey 和预分区OpenTSDB 的查询能力又偏弱。Lindorm TSDB 给我感觉最顺手的点在于它本身是 Serverless 形态的时序引擎写入链路直接吃高并发写入不需要提前规划太多分区细节同时底层是列式存储和 LSM 结构压缩比高长时间保存的成本可控。而且 Lindorm TSDB 的模型很接近 InfluxDB 的 tag 和 field 概念适合描述带多个维度的传感器数据。查询方面支持类 SQL 和 InfluxDB 兼容协议业务侧不用改写太多代码就能直接对接。对于需要长时间保留原始或者降精度数据的监控类场景这几点都命中需求。1.3 Flink 在计算侧的位置既然存储选好了计算这层自然就交给了 Flink。同样是流计算Spark Streaming 是微批模型窗口延迟和状态管理的灵活性都不如 Flink 的纯流式处理。Flink 的事件时间语义、精确一次处理、原生的窗口 API 和丰富连接器生态在处理乱序传感器数据时确实更省心。另外我刻意没有把聚合逻辑下沉到 Lindorm 侧比如说每隔一分钟算一次均值理论上数据库也能做但实时性、窗口灵活性都会受限。Flink 在写入之前把一些高频、实时性要求高的聚合结果先算好Lindorm 只负责存“算好的结果”和“需要保留全量细节的原始数据”。这样计算和存储解耦后续想调整计算逻辑不需要动存储模型。1.4 整体链路长什么样这套方案最终跑通的链路是设备端采集数据 → Kafka → Flink 清洗和窗口聚合 → Lindorm TSDB → 上层查询/大屏/告警。采集端设备或网关将传感器数据上报到 KafkaKafka作为缓冲层和解耦层削峰填谷避免写入抖动直接打到下游Flink做数据格式解析、脏数据过滤、乱序处理、窗口聚合、实时降采样然后写入 LindormLindorm TSDB承接实时流写入和查询请求支撑大屏、报表、告警和回放分析中间加 Kafka 不是为了多一层组件而是为了保证生产端和消费端的速率解耦。设备上报经常有毛刺比如每天固定时刻大量数据同时进来如果没有缓冲层Lindorm 的写入压力会被瞬间顶满。Kafka 加上 Flink 的 checkpoint 机制也让任务重启后的数据回溯变得简单可靠。2. 集成前的关键准备版本、依赖与表模型设计2.1 依赖版本组合怎么选我这边环境是 Flink 1.17Kafka 客户端版本 3.2Lindorm TSDB 使用官方提供的 Java SDK。Flink 版本不用过于追新稳定和生态兼容最重要。1.17 的窗口和水印 API 已经非常成熟社区资料也多遇到问题找答案容易。SDK 的话建议尽量用当前版本对应的最新稳定版避免太旧导致签名或协议不兼容。如果是 Maven 项目核心依赖大致是这样dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version1.17.1/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-kafka/artifactId version1.17.1/version /dependency !-- Lindorm TSDB SDK 按具体版本引入 -- dependency groupIdcom.aliyun/groupId artifactIdlindorm-tsdb-client/artifactId version.../version /dependencyFlink 连接 Lindorm TSDB 这件事没有天然像 Kafka 那么开箱即用的官方连接器大多数项目是基于 TSDB SDK 自己写一个 Sink。这个方式成本不高后续我会单独讲自定义 Sink 的实现细节。如果你刚好用的是 Lindorm 的大数据组件那套生态可能会有现成的封装或配置但通用性最强的做法还是 SDK 自研。2.2 表模型设计tag、field、时间戳时序表模型搞清楚了后面所有计算和查询都顺。Lindorm TSDB 的模型核心是“时间戳 标签 字段”这跟 InfluxDB 一个思路。标签用来描述数据的归属维度字段才是真正要存储的指标值。当时的传感器数据结构大概是这样的{ device_id: DEVICE_001, region: cn-hangzhou, type: temperature, timestamp: 1735689600000, temperature: 36.5, humidity: 70.2, voltage: 12.4 }建表时可以参考类似下面的 SQL 风格CREATE TABLE sensor_data ( device_id VARCHAR(256), region VARCHAR(64), ts BIGINT, temperature DOUBLE, humidity DOUBLE, voltage DOUBLE, PRIMARY KEY(ts, device_id, region) );这里设计标签和字段有一个比较重要的原则需要做筛选和分组操作的数据放到 tag 里例如 device_id、region、type纯指标数据放到 field 里例如 temperature、humidity、voltage。反过来如果把 region 放到 field 里后面按区域查询和聚合会非常痛苦因为你只能扫描全部记录再做过滤。时间戳统一用毫秒。Kafka 里的原始数据可能是秒级或毫秒级单位不统一如果源头是不同厂商的设备尤其要小心。我之前遇到过一次“数据看起来少了一大部分”的诡异问题最后发现就是部分设备上报的时间戳是秒部分又是毫秒写入 Lindorm 后分布完全乱了。TTL 和预分区也值得提前思考。原始数据一般设置相对短一些的保留周期比如 7 天聚合结果数据可以保留更久比如 30 天甚至更长时间。分区策略上Lindorm TSDB 有能力自动处理但我建议还是根据时间范围和吞吐做基础规划尤其是写入峰值非常高的场景。2.3 时间语义与乱序容忍度Flink 里面最影响计算结果的就是时间语义选择。用处理时间虽然简单但聚合出来的窗口结果会随着任务处理速度变化产生偏移对监控类场景来说不可接受。我全部采用事件时间也就是以传感器采集端产生的时间为准。事件时间需要配套水印机制来容忍网络和采集抖动。传感器数据的乱序程度通常不会特别大几十秒的迟到算正常个别设备断网重连后可能晚几分钟补发。我用的水印策略是固定乱序容忍度允许迟到 10 秒超过窗口末尾一定时间的数据不再参与窗口聚合而是扔到侧输出流里单独处理。这样做既保证绝大多数数据能进对窗口又不让个别异常迟到数据拖住整个窗口的触发。3. 核心代码从 Kafka 到 TSDB 的完整链路实现3.1 全局代码框架先搭主链路骨架代码看着不多但每个环节都有必要单独抽出来。主流程大致是读 Kafka、解析 JSON、分配时间戳和水印、按维度分组、开窗口聚合并写入 Lindorm。StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(4); env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE); DataStreamString rawStream env.addSource(new FlinkKafkaConsumer( sensor_topic, new SimpleStringSchema(), kafkaProps )).setStartFromLatest(); DataStreamSensorRaw parsed rawStream .map(new JsonToSensorRaw()) .filter(Objects::nonNull); DataStreamSensorAgg aggStream parsed .assignTimestampsAndWatermarks( WatermarkStrategy.SensorRawforBoundedOutOfOrderness(Duration.ofSeconds(10)) .withTimestampAssigner((row, ts) - row.timestamp) ) .keyBy(r - r.deviceId _ r.region) .window(TumblingEventTimeWindows.of(Time.minutes(1))) .aggregate(new SensorAggFunc()); aggStream.addSink(new LindormTsdbSink());并行度设置这里我的经验是不要盲目调大。Sink 是写外部存储的如果 Flink 端并行度太高Lindorm 的写入压力会成倍增加。主线并行度控制在 4 到 8 之间配合后面的批量参数整体吞吐和写入稳定性是最好的。3.2 数据解析与水印JSON 解析看似简单但坑往往在脏数据上。设备偶尔会上报空值、缺字段、字符串和数字混用的数据。我在 map 函数里统一做防御任何解析异常直接返回 null由后续的 filter 过滤掉。不要因为一条脏数据就导致整个任务失败。水印这块需要懂一个点水印不是无限往后推进的。如果某个 source 分区长时间没数据它对应的水印可能一直停在旧位置整个窗口都不触发。这种情况要观察 Flink UI 里的水印是否在持续增长。如果真的发现某个设备的 topic 分区没数据优先检查上游生产端而不是 Flink。3.3 窗口聚合与降精度在数据落库前做一次分钟级的聚合好处非常明显Lindorm 里存储的数据量直接降低一个数量级。以 1 分钟为窗口同一个设备、同一分钟内的多跳数据聚合成一条记录包含平均温度、最高温度、最低温度、采集点数等指标。窗口聚合用的增量聚合方式不必等到窗口闭合才计算每来一条数据都会更新累加器内存开销非常稳定。对于一小时甚至一天级别的窗口这个优势更明显。我每个窗口还会额外输出一个 count 字段方便排查是不是存在某个窗口内数据量明显偏少的情况。迟到数据不要丢弃侧输出流单独接收。我配置了允许 5 分钟的迟到时间超过 5 分钟的数据进入侧输出流之后另写任务做补偿写入或者人工分析。如果不做这一层兜底上游偶发断网补发时会发现数据缺失。3.4 自定义 Lindorm Sink 的关键实现这是整个集成里最值得细写的部分。当时我选择基于 SDK 自己封装 RichSinkFunction核心原因是可以完全控制写入频率、批量大小和重试策略。代码骨架如下public class LindormTsdbSink extends RichSinkFunctionSensorAgg { private transient TsdbClient client; private transient ListPoint batch; private int batchSize 500; private long lastFlushTime 0L; private long flushIntervalMs 2000L; Override public void open(Configuration parameters) throws Exception { client TsdbClientBuilder.create() .host(lindormHost) .port(lindormPort) .database(sensor_db) .build(); batch new ArrayList(batchSize); } Override public void invoke(SensorAgg value, Context context) throws Exception { Point point Point.builder() .measurement(sensor_agg) .tag(device_id, value.deviceId) .tag(region, value.region) .timestamp(value.windowEnd) .field(avg_temperature, value.avgTemp) .field(max_temperature, value.maxTemp) .field(min_temperature, value.minTemp) .field(count, value.count) .build(); batch.add(point); if (batch.size() batchSize || System.currentTimeMillis() - lastFlushTime flushIntervalMs) { flush(); } } private void flush() throws Exception { if (batch.isEmpty()) return; client.write(batch); batch.clear(); lastFlushTime System.currentTimeMillis(); } Override public void close() throws Exception { if (batch ! null !batch.isEmpty()) { flush(); } if (client ! null) { client.close(); } } }这个 Sink 的逻辑是“攒批 定时刷出”。满足条件才真正写 Lindorm而不是每来一条数据就网络请求一次。传感器场景每分钟聚合后的数据量不算特别大这种攒批写法的吞吐足够而且能显著降低外部存储的连接压力。需要特别强调的是失败重试。写入 Lindorm 可能因为瞬时限流或者网络抖动失败我在 flush 里加了简单重试重试三次间隔指数退避。如果连续失败就抛异常触发 Flink 的 checkpoint 机制任务自动重启已攒批的数据会重新发给 Lindorm。因为写入是基于 upsert 语义的重复写同一时间点和同一标签的数据只会覆盖旧值不会产生重复脏数据。3.5 数据一致性语义的选择这里做一个简单取舍。理论上如果想要精确一次得用 Flink 的两阶段提交协议但 TSDB 这类系统通常不直接支持分布式事务协调。好在时序数据天然是幂等覆盖写时间戳加标签就可以唯一确定一条记录重复写入不会累积出错误数据。所以我最终选择的是至少一次语义结合 Lindorm 的 upsert 特性效果上等于精确一次。这一点在业务层面非常重要但实现代价几乎为零。如果你对接的目标存储不支持幂等写那就要认真考虑乱序导致重复数据的问题了。4. 上线之后的坑与排查记录4.1 时间戳差八小时有一次大屏上查出来的曲线总是慢八个小时一开始我怀疑是 Flink 窗口没闭合排查了半天发现是 Kafka 里的时间戳在有部分场景下被上游转换成了 UTC 字符串解析后变成标准时区而 Lindorm 查询端默认用的本地时区结果整体错位 8 小时。这个坑最终靠统一数据链路规范解决所有设备上报的时间戳一律转成 Unix epoch 毫秒在 Kafka 消息里用 long 类型字段传输Flink 解析后直接使用。如果非要用字符串时间必须带上时区信息否则不可能避免这种误差。4.2 写入限流导致 Sink 长期背压上线后某段时间 Flink UI 里 Sink 的 backPressure 一直飘红任务延迟越来越大。排查后发现是 Lindorm 的写入容量被打满限制含义就是单分区或单表写入速率达到阈值后会被服务端拒绝。原因就是上游某个活动期间数据量翻了几倍Sink 的攒批参数没调每批次写入比之前更大触发了服务端的限流。解决方式是同时做了三件事把 batchSize 调小一点从 500 改成 200减少单次写入的峰值请求体量把 flush interval 从 2 秒改成 1 秒让数据更均匀地写出去然后适当降低 Sink 的并行度避免多个线程同时猛轰。调优之后背压稳定在低水位。4.3 水印不前进导致窗口迟迟不触发Flink 事件时间最容易出问题的就是这个现象数据一直在进来但窗口就是不算看起来好像没有结果写出。后来查原因是 Kafka 的某个分区对应的并行实例一直没有数据由于水印的计算是取所有输入分片的最小水印这个没有数据的分片把整体水印卡住了。这种情况下如果任务重启读取策略还设成了从最新开始消费那没有数据的这个分片就一直取不到起始位置。我想的解决方法是把 Kafka 消费的起始位置改成最早并给 source 指定一个多个分片都能推进的策略保证所有分片都有水印输出。除此之外空闲分片超时机制也建议开启Flink 1.17 里可以设置空闲流时间超过阈值就直接忽略这个分片的水印影响。这个参数对多分区 Kafka 源非常关键。4.4 单设备数据倾斜设备上报频率不是均匀的可能某个区域的 10 台设备占了整体流量的一半这就导致按 device_id 做 keyBy 之后某个子任务负载巨大其他子任务空闲。时序场景里数据倾斜通常不需要像通用计算那样加随机前缀再二次聚合因为加随机前缀会破坏同一设备的时间序列连续性。我采用的缓解措施是自然维度拆分把 key 从纯 device_id 改成 device_id region让分组更细同时在生产端控制同一设备的写入速率峰值。通过这两种方式任务负载基本能够平衡。4.5 连接空闲断开这个问题发生在任务长时间空转、数据量很低的时段。Lindorm 服务端的空闲连接可能被断开Sink 里的连接对象还以为是有效的下一条数据写入时直接抛连接异常。我最后的处理方式是给连接加一个定期校验和重建机制定时任务里每 5 分钟检查一次连接状态必要时重新创建实例。同时在写入异常里加了一个判断逻辑如果异常类型是连接关闭就重建连接后重试一次。这个方法并不复杂但很实用。4.6 问题和解决方案速查问题现象可能原因解决方向查询结果整体偏移数小时时间戳单位或时区不一致全链路统一为 Unix epoch 毫秒Sink 背压高任务延迟大Lindorm 容量打满调小批量、降并行度、分散写入窗口长时间不触发空闲分区卡住水印设置空闲分片超时开启分流推进单子任务负载过高key 不均匀扩展分组维度细化 key低流量时段写入报连接异常服务端空闲断开定期重建连接异常重试窗口内数据量忽高忽低上游设备上报抖动侧输出流接住迟到数据另做补偿5. 资源规划、成本与进一步优化5.1 并发度与写入吞吐的匹配Flink 并行度不能拍脑袋定。我一般按“Kafka topic 分区数”来设 source 并行度比如 topic 设置了 8 个分区source 并行度就是 8。窗口和 Sink 的并行度不一定要跟着 8 走可以设小一些避免每个计算子任务都独立发连接。Sink 并行度通常设置为 1 到 4保证批量够大、写入够顺。Lindorm 侧的容量规划可以按峰值 QPS 来预估。假设每台设备 10 秒上报一条记录10 万台设备峰值每秒是 1 万条写入聚合后分钟级输出只有几千条每秒存储压力主要是原始数据。如果原始数据只保留 7 天字段还不算太多单节点可以用很久。5.2 降本与降精度策略这套链路里浪费最多的是原始数据全量存。后来我调整成双重存储策略原始数据降精度存比如 10 秒一条落库同时窗口聚合的分钟级结果单独存一张表。查询大屏和告警直接用分钟级聚合数据需要分析原始波形时才去查原始表。这样把存储成本压低了一个数量级查询响应也没有变慢。TTL 同样要利用起来。不同用途的表设置完全不同的保留周期聚合结果和原始数据分开设置避免一张表里既有短期数据又有长期数据否则运维上的复杂度会很高。5.3 监控与告警配套整条链路不能只看业务跑通运行状态也要可视化监控。Flink UI 的水印、backPressure 和 checkpoint 时长是必看的。把 Flink 的自定义指标接入 Prometheus 之后我把 Sink 的写入失败次数、批量 flush 耗时、Lindorm 写入响应码这些指标都建了告警。实操中发现一个很小的指标反而是最关键的批量积压条数。如果这个值持续升高说明 Lindorm 的写入吞吐跟不上 Flink 的下发速度就算背压还没飘红也已经在逼近瓶颈了。把这个指标配上告警能比背压更早发现潜在问题。6. 最后的几点经验这套 Flink 与 Lindorm TSDB 的集成链路跑了大半年整体稳定但我最大的体会是技术选型只是开始真正决定成败的是细节。时间戳统一、表模型设计、批量参数调优、水印和延迟数据处理每一个环节不提前考虑到线上就会还回来。如果有人现在让我重新搭一次我第一件事一定把全链路时间戳单位写死成毫秒并让所有生产端强制遵守。数据模型一旦因为时间基准不统一而乱了后面所有清洗和聚合逻辑都会产生系统性偏差那时候再来修就非常痛苦了。再提一个小技巧Sink 的 flush 方法里加一个耗时打点写入耗时突然升高通常意味着 Lindorm 侧在 compaction 或扩容这个信号比服务端报错更早。观测到这个现象时适当降低写入频率可以让服务端更平滑地度过资源调整期避免硬碰硬触发限流。希望这些实践经验能帮你少走一些弯路。