1. Flink 数据读取的整体设计思路1.1 为什么数据读取是 Flink 作业的第一道生死关很多人刚接触 Flink 的时候注意力都放在窗口、状态、Checkpoint 这些高级机制上觉得数据读取无非就是调个readText或者addSource能跑就行。我带过几个新人几乎每个人都在这个环节栽过跟头。实际情况是数据读取决定了整个作业的并行度上限、吞吐瓶颈、以及容错语义的起点。Source 这一层没设计好后面算子写得再漂亮也是白搭。举个最直观的例子你从消息队列里读数据如果 Source 并行度和分区数对不上就会出现部分 SubTask 空转、部分 SubTask 被打爆的情况整个作业的吞吐被最慢的那个并行子任务拖死。再比如你没有正确处理 Source 的偏移量提交作业一重启就从头消费下游数据库直接被重复数据冲垮。这些都是真实生产里血淋淋的教训。所以我把数据读取放在 Flink 基础系列的第一篇不是因为它简单而是因为它太关键了。这一篇我会把 Flink 里几种主流的数据读取方式掰开揉碎讲清楚从最基础的集合、文件读取到生产环境最常用的 Kafka 读取再到自定义 Source 的写法以及并行度、水位线、容错这些绕不开的坑。1.2 Flink 数据源的分层结构在动手写代码之前得先搞清楚 Flink 的数据源在架构上是怎么分层的。理解了这个你才知道每种 Source 的适用边界在哪。Flink 的数据源大致可以分成这么几类内置 Source框架自带的比如fromCollection、readTextFile、fromElements、socketTextStream这些。适合测试和简单场景生产环境基本不用。连接器 Source通过官方或第三方 Connector 接入外部系统比如 Kafka、Kinesis、RabbitMQ、JDBC、Elasticsearch 等。这是生产环境的主力。自定义 Source继承SourceFunction或实现新的Source接口自己控制数据的产生和发送逻辑。适合接入那些没有现成连接器的系统。从 Flink 1.12 开始社区在推动一套新的 Source APIFLIP-27目的是统一批流的数据读取接口把并行度发现、分片分配、水位线生成这些职责从 SourceFunction 里解耦出来。到了 Flink 1.14 之后Kafka Connector 已经全面切换到新 API。这个演进方向很重要因为它直接影响到你选哪个版本的连接器、怎么写自定义 Source。提示如果你现在还在用SourceFunction写生产级 Source建议尽快了解新的Source接口。旧接口在并行度动态发现、批流统一这些方面有明显短板社区后续的维护重心已经转移了。1.3 选型时到底该看哪几个维度面对一个具体的读取需求怎么选 Source我一般看四个维度第一数据源类型和接入成本。如果数据在 Kafka 里那没什么好纠结的直接用 Kafka Connector。如果数据在一个冷门的自研系统里没有现成连接器那就得自己写。第二容错语义要求。你的作业要求 At-Least-Once 还是 Exactly-OnceKafka Connector 配合 Checkpoint 可以做到 Exactly-Once但你自己写的 Source 如果偏移量管理没做好可能连 At-Least-Once 都保证不了。第三并行度和吞吐。数据源本身支不支持并行读取比如 Kafka 的并行度受分区数限制文件读取可以按文件切分JDBC 读取就得考虑怎么分片查询。第四水位线和事件时间。如果你的作业要用事件时间处理Source 必须能产生水位线。Kafka 里每条消息自带时间戳处理起来比较自然自定义 Source 就得自己决定怎么发水位线。把这四个维度想清楚选型基本就不会跑偏。2. 核心数据读取方式的细节拆解2.1 集合与元素读取测试阶段的快车道fromCollection、fromElements、fromParallelCollection这几个方法是写单元测试和本地调试时的好帮手。它们的共同特点是把内存里的数据直接变成 DataStream不涉及外部系统启动快、无依赖。StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); // 从集合读取 DataStreamString fromColl env.fromCollection(Arrays.asList(a, b, c)); // 从元素直接读取 DataStreamInteger fromElems env.fromElements(1, 2, 3, 4, 5); // 并行集合可以指定切分方式 DataStreamLong fromParallel env.fromParallelCollection( new NumberSequenceIterator(1L, 100L), Long.class );这里有个细节很多人不知道fromCollection默认是单并行度的也就是说不管你env设了多少并行度这个 Source 只会用一个线程读。数据量小的时候无所谓但如果你拿它做压测就会误以为整个作业的吞吐就这么点。fromParallelCollection才是可以真正并行读取的版本它通过NumberSequenceIterator或LongValueSequenceIterator把序列切分到各个 SubTask。注意这几个方法都只在本地测试用别拿到生产环境。它们的数据全在 JobManager 内存里数据量一大直接 OOM而且没有任何容错能力。2.2 文件读取批处理场景的常青树文件读取在 Flink 里有两套 API老的是readTextFile、readFile新的是FileSource。老 API 在批流统一之前用得很多现在新项目建议直接用FileSource。// 老 API简单但不支持复杂格式 DataStreamString text env.readTextFile(hdfs://path/to/file); // 新 API功能完整 FileSourceString source FileSource .forRecordStreamFormat(new TextLineInputFormat(), new Path(hdfs://path/to/dir)) .monitorContinuously(Duration.ofSeconds(10)) .build(); DataStreamString stream env.fromSource( source, WatermarkStrategy.noWatermarks(), file-source );文件读取的核心难点在于目录监控和文件切分。monitorContinuously让 Source 定期扫描目录发现新文件就读取这个在流式场景里很有用。文件切分则是把一个大文件按块分给不同 SubTask 并行读提升吞吐。这里有个坑文件读取的并行度不是你想设多少就设多少。如果目录下只有一个文件而且这个文件不支持切分比如压缩格式那实际并行度就是 1。所以生产环境里文件源通常要配合多文件、可切分格式来用。2.3 Kafka 读取生产环境的主力选手Kafka 是 Flink 生产环境里出现频率最高的数据源没有之一。它的 Connector 经历了从FlinkKafkaConsumer到KafkaSource的演进现在新项目一律用KafkaSource。KafkaSourceString source KafkaSource.Stringbuilder() .setBootstrapServers(broker1:9092,broker2:9092) .setTopics(topic-a, topic-b) .setGroupId(flink-group-001) .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST)) .setValueOnlyDeserializer(new SimpleStringSchema()) .setProperty(partition.discovery.interval.ms, 60000) .build(); DataStreamString stream env.fromSource( source, WatermarkStrategy.StringforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, ts) - extractTimestamp(event)), kafka-source );这段代码里有几个参数值得单独拎出来说setStartingOffsets决定了作业第一次启动或者没有可提交偏移量时从哪开始读。committedOffsets是默认值优先用消费者组提交的偏移量没有的话按OffsetResetStrategy走。earliest和latest是另外两个常用选项。这里的选择直接影响你作业重启后的行为选错了要么丢数据要么重复消费。partition.discovery.interval.ms是动态分区发现。Kafka 里新增了分区如果不开启这个Flink 是感知不到的新分区的数据就一直积压。生产环境建议开启间隔设个 30 秒到 1 分钟都行。setGroupId这个参数在新 API 里其实只用于偏移量提交和监控不再像老 API 那样决定消费的分配。Flink 的 Kafka Source 是自己管理分区分配的不依赖 Kafka 的消费者组协调机制。2.4 自定义 Source没有连接器时的兜底方案当你要读的数据源没有现成连接器时就得自己写 Source。新 API 下核心是实现Source、SplitEnumerator、SourceReader这三个接口说实话门槛不低。如果只是简单场景用SourceFunction也能凑合但要清楚它的局限。public class SimpleSource extends RichSourceFunctionString { private volatile boolean running true; Override public void run(SourceContextString ctx) throws Exception { while (running) { String data fetchFromExternal(); synchronized (ctx.getCheckpointLock()) { ctx.collect(data); } Thread.sleep(100); } } Override public void cancel() { running false; } }这段代码里最关键的是synchronized (ctx.getCheckpointLock())。它的作用是保证collect和 Checkpoint 的快照操作不会并发执行否则可能出现数据已经发出去了但快照里没记录偏移量的情况导致重启后重复消费。这个锁是SourceFunction容错的核心很多人写自定义 Source 时漏掉它结果 Exactly-Once 直接失效。3. 实操过程与关键环节实现3.1 从零搭建一个 Kafka 读取作业光看代码不够我带你走一遍完整的搭建流程。假设我们要做一个实时读取 Kafka 订单数据、解析后打印的作业。第一步引入依赖。Maven 里加上 Kafka Connectordependency groupIdorg.apache.flink/groupId artifactIdflink-connector-kafka/artifactId version3.1.0-1.18/version /dependency版本号要和你的 Flink 版本对应。Flink 1.18 对应 Connector 3.1.01.17 对应 3.0.x这个对应关系搞错了会报各种NoSuchMethodError。第二步配置 Source。除了前面提到的基础参数生产环境还要考虑KafkaSourceString source KafkaSource.Stringbuilder() .setBootstrapServers(brokers) .setTopics(order-topic) .setGroupId(order-consumer) .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST)) .setValueOnlyDeserializer(new SimpleStringSchema()) .setProperty(partition.discovery.interval.ms, 30000) .setProperty(commit.offsets.on.checkpoint, true) .setProperty(fetch.max.bytes, 52428800) .build();commit.offsets.on.checkpoint设为 true 后偏移量会随 Checkpoint 一起提交到 Kafka这样即使 Flink 的状态后端丢了也能从 Kafka 恢复消费位置。第三步设置水位线策略。如果作业要用事件时间这一步不能省WatermarkStrategyString watermarkStrategy WatermarkStrategy .StringforBoundedOutOfOrderness(Duration.ofSeconds(10)) .withTimestampAssigner((event, recordTs) - { JSONObject obj JSON.parseObject(event); return obj.getLongValue(eventTime); }) .withIdleness(Duration.ofMinutes(1)); DataStreamString stream env.fromSource(source, watermarkStrategy, kafka-source);withIdleness是处理空闲分区的关键。如果某个 Kafka 分区长时间没数据它的水位线不推进会拖住整个作业的事件时间。设置空闲超时后Flink 会把这个分区标记为空闲不再参与水位线计算。第四步验证并行度。作业的 Source 并行度不能超过 Kafka 分区数否则多出来的 SubTask 会空转。可以在提交时用-p参数指定或者在代码里env.setParallelism()。3.2 偏移量提交与容错配置的实操细节偏移量管理是 Kafka 读取里最容易出问题的地方。我见过太多作业因为偏移量配置不当要么重复消费要么丢数据。Flink 的 Kafka Source 有两种偏移量提交方式提交方式触发时机适用场景风险Checkpoint 提交Checkpoint 完成时要求 Exactly-Once依赖 Checkpoint 正常自动提交定时提交对一致性要求低可能重复或丢失生产环境我强烈建议用 Checkpoint 提交。配置方式就是前面提到的commit.offsets.on.checkpointtrue同时确保 Checkpoint 正常开启env.enableCheckpointing(60000); env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000); env.getCheckpointConfig().setCheckpointTimeout(120000); env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);Checkpoint 间隔设多少合适这个要看你的数据量和容错要求。间隔太短Checkpoint 开销大间隔太长重启后要重放的数据多。一般 1 分钟是个比较稳妥的起点数据量大的话可以放宽到 3 到 5 分钟。提示Checkpoint 提交偏移量时偏移量是Checkpoint 触发那一刻的位置不是Checkpoint 完成那一刻的位置。所以即使 Checkpoint 失败偏移量也不会错乱这是 Flink 两阶段提交的精髓。3.3 自定义 Source 的完整实现与容错假设我们要从一个自研的 HTTP 接口拉数据接口支持按游标分页。这种场景没有现成连接器得自己写。public class HttpCursorSource extends RichSourceFunctionString { private volatile boolean running true; private String cursor; private transient CloseableHttpClient httpClient; Override public void open(Configuration parameters) throws Exception { httpClient HttpClients.createDefault(); // 从状态恢复游标 if (getRuntimeContext().getIndexOfThisSubtask() 0) { // 只有第一个 SubTask 负责拉取 } } Override public void run(SourceContextString ctx) throws Exception { while (running) { ListString batch fetchBatch(cursor); if (batch.isEmpty()) { Thread.sleep(1000); continue; } synchronized (ctx.getCheckpointLock()) { for (String record : batch) { ctx.collect(record); } cursor extractNextCursor(batch); } } } Override public void cancel() { running false; } Override public void close() throws Exception { if (httpClient ! null) { httpClient.close(); } } }这个实现有几个要点游标状态要持久化。上面代码里游标只存在内存里作业重启就丢了。正确做法是用ListState或OperatorState把游标存起来在snapshotState里写、在initializeState里读。并行度要谨慎。HTTP 接口通常不支持并行拉取所以这个 Source 的并行度应该设为 1。如果非要并行得让接口支持按分片拉取每个 SubTask 负责一个分片。背压要处理。如果下游处理慢collect会阻塞这时候 Source 的拉取循环也会被卡住。这是正常的背压机制不用刻意规避但要注意别在collect外面做太多耗时操作。4. 常见问题与排查技巧实录4.1 数据读取环节的高频故障速查下面这张表是我这些年踩坑总结出来的基本覆盖了 80% 的读取问题现象可能原因排查方向解决方式作业启动后不消费数据起始偏移量设为 latest 且无新数据检查setStartingOffsets改为 earliest 或确认有新数据部分分区数据积压并行度小于分区数对比 Source 并行度和分区数提高并行度至等于分区数重启后重复消费偏移量未提交或提交失败检查 Checkpoint 是否正常开启 Checkpoint 提交偏移量水位线不推进存在空闲分区查看各分区数据速率设置withIdleness反序列化报错数据格式与 Schema 不匹配打印原始数据查看换用合适的 Deserializer吞吐上不去单并行度或反序列化慢看各 SubTask 的吞吐指标提高并行度、优化反序列化4.2 反序列化的性能陷阱反序列化看起来是个小环节但它经常成为吞吐瓶颈。我做过一个测试同样的 Kafka 数据用SimpleStringSchema和自定义的 JSON 反序列化吞吐差了将近 40%。问题出在很多人在deserialize方法里做了太多事解析 JSON、做字段校验、甚至查数据库。这些操作应该尽量后移到map或process算子让 Source 只负责把字节转成对象。// 不推荐在反序列化里做重活 public class HeavyDeserializer implements DeserializationSchemaOrder { Override public Order deserialize(byte[] message) { JSONObject obj JSON.parseObject(message); // 一堆校验和转换逻辑 return convert(obj); } } // 推荐反序列化只做最轻的转换 public class LightDeserializer implements DeserializationSchemaString { Override public String deserialize(byte[] message) { return new String(message, StandardCharsets.UTF_8); } }另外DeserializationSchema的isEndOfStream方法在流式场景里永远返回 false别在里面写逻辑。4.3 并行度与分区数的匹配问题这是新手最容易忽略的点。Kafka Source 的并行度如果大于分区数多出来的 SubTask 会一直空闲白白占用资源如果小于分区数一个 SubTask 要消费多个分区吞吐受限。正确的做法是让 Source 并行度等于分区数。但这里有个细节Flink 的并行度是全局设置的如果 Source 并行度是 4下游算子也会是 4。如果下游不需要这么高并行度可以在 Source 后面加rebalance或者单独设置算子的并行度。DataStreamString stream env.fromSource(source, watermarkStrategy, kafka-source) .setParallelism(8); // Source 并行度设为 8 stream.map(...).setParallelism(4); // 下游算子并行度设为 44.4 水位线相关的疑难杂症水位线不推进是流式作业里最让人头疼的问题之一。常见原因有这么几个空闲分区拖累。前面提过用withIdleness解决。时间戳提取失败。如果extractTimestamp返回了负数或者异常值水位线会乱掉。建议在提取时加个兜底.withTimestampAssigner((event, recordTs) - { try { long ts parseTimestamp(event); return ts 0 ? ts : System.currentTimeMillis(); } catch (Exception e) { return System.currentTimeMillis(); } })乱序程度设置过大。forBoundedOutOfOrderness设得太大水位线推进慢窗口迟迟不触发。设得太小迟到数据被丢弃。这个值要根据实际数据的乱序情况来定一般取 P99 的乱序延迟。多个 Source 的水位线对齐。如果作业有多个 SourceFlink 会取所有 Source 水位线的最小值。任何一个 Source 卡住整个作业的事件时间就停了。这种情况要么保证所有 Source 都活跃要么用withIdleness让空闲 Source 退出水位线计算。5. 数据读取的进阶优化思路5.1 大吞吐场景下的读取调优当单条数据不大、但 QPS 极高时读取环节的调优重点在批量拉取和减少网络往返。Kafka Source 有几个参数值得调fetch.min.bytes每次拉取的最小字节数设大一点可以减少请求次数但会增加延迟。fetch.max.wait.ms拉取等待时间配合fetch.min.bytes使用。max.partition.fetch.bytes每个分区单次拉取的最大字节数。receive.buffer.bytes接收缓冲区大小。这些参数没有万能值要根据数据特征压测。我的经验是先把fetch.min.bytes设到 1MB 左右fetch.max.wait.ms设 500ms然后根据延迟和吞吐的平衡微调。5.2 数据倾斜的识别与缓解数据倾斜在读取环节的表现是某些 SubTask 的numRecordsIn远高于其他 SubTask。原因通常是 Kafka 分区之间数据量不均或者某个 key 的数据特别集中。识别方法很简单看 Flink Web UI 里各 SubTask 的指标或者用 Metrics 上报到监控系统。缓解手段有调整分区策略如果是自己写数据到 Kafka确保 key 的哈希分布均匀。Source 后加 rebalance把数据重新均匀分配但会引入网络开销。两阶段聚合如果倾斜发生在聚合算子可以在 Source 后先做一次本地聚合。5.3 读取环节的监控指标生产环境一定要把读取环节的指标监控起来否则出了问题只能靠猜。核心指标包括numRecordsIn各 SubTask 的输入记录数看倾斜。numRecordsInPerSecond输入速率看吞吐。currentFetchEventTimeLagKafka 消费延迟看积压。currentEmitEventTimeLag数据从产生到发出的延迟。checkpointDurationCheckpoint 耗时看容错开销。这些指标可以通过 Flink 的 Metrics Reporter 上报到 Prometheus 或 InfluxDB再配 Grafana 看板。我一般会设几个告警消费延迟超过阈值、Checkpoint 连续失败、SubTask 之间吞吐差异超过 3 倍。6. 我在实际项目中的几点体会写了这么多最后分享几个只有真正在生产环境跑过作业才会有的体会。第一Source 的并行度宁大勿小。我早期做过一个作业Source 并行度设得刚好等于分区数结果后来 Kafka 扩容加了分区作业没跟着调整新分区的数据一直积压。后来我养成了习惯Source 并行度留一点余量配合动态分区发现扩容时不用改作业。第二起始偏移量的选择要结合业务。earliest和latest没有绝对的好坏。做实时监控用latest避免重启时重放历史数据做数据同步用earliest保证不丢数据。关键是团队里要有共识别今天这个人改成latest明天那个人改成earliest。第三自定义 Source 能不用就不用。除非真的没有现成连接器否则别自己造轮子。自定义 Source 的容错、并行度、水位线这些都要自己处理坑太多。我见过一个团队为了读一个冷门数据库自己写了 Source结果偏移量管理有 bug上线三个月才发现数据一直在重复。第四测试环境一定要模拟真实数据量。很多读取问题在小数据量下根本暴露不出来。我建议在测试环境用生产环境的采样数据把 QPS 压到生产的一半以上跑个几小时看看有没有积压、倾斜、Checkpoint 超时。第五版本兼容性要提前确认。Flink 和 Connector 的版本对应关系很严格升级 Flink 时一定要同步升级 Connector。我踩过一次坑Flink 从 1.13 升到 1.14Kafka Connector 没跟着升结果作业启动就报ClassNotFoundException排查了半天才发现是版本问题。数据读取这块内容其实还有很多可以展开的比如 Flink SQL 里的 Connector 配置、CDC 场景下的增量读取、以及和 Pulsar、Kinesis 这些消息系统的集成。后面有机会再单独写。如果你正在做 Flink 的数据接入希望这篇能帮你少走点弯路。