1. 为什么数据读取是Flink作业的第一道生死关很多人刚接触Flink的时候注意力几乎全放在窗口计算、状态管理、Exactly-Once语义这些高级话题上觉得数据读取无非就是调一个addSource或者fromSource的事能跑通就行。我一开始也是这么想的直到有一次帮朋友排查一个线上作业才发现数据读取这一层埋的坑远比想象中多得多。那个作业的逻辑很简单从消息队列读数据做一轮清洗再写到下游存储。代码本身没几行但作业跑起来之后延迟忽高忽低高峰期还会反压到整个链路卡死。折腾了大半天最后定位到的根因居然是最不起眼的Source并行度和分区策略没配对——Source端并行度设成了1下游算子开了8个并行度数据全堵在入口那一个SubTask里后面的算子再快也白搭。这件事让我意识到一个很朴素的道理Flink作业的性能上限往往在数据读取这一层就已经被决定了。Source是整条数据流水线的水龙头水龙头开多大、怎么开、开几个直接决定了后面所有环节能拿到多少水。你后面窗口算得再精妙、状态后端调得再顺滑入口堵住了一切都是空谈。所以这篇内容我想把Flink基础之数据读取这件事掰开揉碎讲清楚。它适合两类人一类是刚上手Flink、还在搞明白Source到底怎么写的新手另一类是写过一些作业、但总觉得数据读取差点意思、想系统梳理一遍的开发者。我会从Flink数据读取的几种典型方式讲起把每种方式的适用场景、核心参数、常见坑点都过一遍中间穿插我自己踩过的坑和总结出来的经验。读完你应该能做到拿到一个数据读取需求能快速判断该用哪种Source、并行度怎么设、参数怎么调而不是凭感觉瞎试。在正式展开之前先明确一个概念边界。Flink里的数据读取通常指两件事一是从外部系统把数据接进来也就是Source端二是数据在Source内部如何被切分、分配、反序列化这涉及到分片策略、并行度、水位线生成等细节。这两件事经常被混为一谈但排查问题时必须分开看——前者是接不接得到后者是接得好不好。后面我会按这个思路来组织内容。2. Flink数据读取的四种典型姿势与选型逻辑Flink发展到现在数据读取的方式经历过几轮演进。早期大家用得最多的是SourceFunction接口后来有了Source的新API再后来社区又推出了DataStream API配合各种Connector的写法。不同版本、不同场景下选择是不一样的。这一章我把常见的四种姿势梳理一遍重点讲清楚每种方式为什么存在以及什么时候该用它。2.1 自定义SourceFunction最灵活但也最容易写错SourceFunction是Flink最早提供的数据源接口它的核心方法就两个run()和cancel()。run()方法里你负责把数据一条条collect出去cancel()方法里负责在作业取消时清理资源。看起来很简单但真正写好一个健壮的SourceFunction并不容易。我见过太多人写SourceFunction时犯同一个错误在run()方法里写了一个while(true)循环然后直接在里面阻塞式地读数据完全不管cancel()。结果作业要停的时候cancel()被调用了但run()里的循环还在傻等作业半天停不下来只能强杀。正确的做法是在循环条件里检查一个volatile boolean标志位cancel()里把它置为false让run()能优雅退出。另一个高频坑点是线程安全。SourceFunction的run()方法是在一个独立线程里跑的如果你在里面访问了外部可变状态又没有做同步很容易出问题。我一般建议把Source内部的状态都收敛到run()方法的局部变量里实在需要跨方法共享的用AtomicBoolean这类并发安全的类型。public class SimpleSource extends RichSourceFunctionString { private volatile boolean running true; Override public void run(SourceContextString ctx) throws Exception { while (running) { String data fetchData(); if (data ! null) { synchronized (ctx.getCheckpointLock()) { ctx.collect(data); } } } } Override public void cancel() { running false; } }上面这段代码里有个细节值得说ctx.collect()为什么要包在synchronized (ctx.getCheckpointLock())里因为Flink做Checkpoint的时候需要保证当前快照对应的数据偏移量和已经collect出去的数据是一致的。如果你在collect的时候不加锁可能出现Checkpoint已经记录了偏移量、但数据还没collect完的情况恢复时就会丢数据或者重复。这个锁是Flink官方推荐的写法很多人不知道写出来的Source在开启Checkpoint后就会出问题。提示自定义SourceFunction适合那些没有现成Connector、数据来源比较特殊的场景比如从一个内部HTTP接口轮询拉数据。但凡是有官方Connector的优先用Connector别自己造轮子。2.2 基于Connector的读取生产环境的首选实际生产里绝大多数数据读取都是通过Connector完成的。Kafka、文件系统、数据库、消息队列Flink社区和各大厂商都提供了成熟的Connector。用Connector的好处是并行度自动适配、Checkpoint集成好了、故障恢复逻辑也帮你处理了你只需要配置参数就行。以最常见的消息队列读取为例Connector的核心配置项其实就那么几个但每一个都直接影响作业行为。我用一个表格把关键参数和它们的影响梳理一下参数作用设置不当的后果消费组ID标识一组消费者多个作业用同一个ID会互相抢分区起始偏移量从哪开始读设错会漏读或重复读大量数据分区发现间隔动态发现新分区设太大新分区数据延迟高设太小增加协调开销水位线策略决定事件时间推进设错会导致窗口不触发或延迟触发空闲分区超时处理无数据分区不设会导致水位线卡住窗口永不触发这里面我想重点说水位线策略和空闲分区超时这两个因为它们最容易被忽略但出问题时又最难排查。水位线策略决定了什么时候认为某个时间点之前的数据都到齐了。如果你用的是事件时间语义水位线设得太激进会丢迟到数据设得太保守窗口迟迟不触发下游看不到结果。我的经验是如果数据乱序程度不高用固定延迟就行延迟大小根据实际乱序情况定一般设成P99乱序时长的1.5倍左右比较稳妥。空闲分区超时这个参数坑过我不止一次。场景是这样的你的数据源有8个分区但某一时刻只有3个分区有数据另外5个是空的。Flink的水位线是取所有分区水位线的最小值那5个空分区的水位线一直不推进导致整个作业的水位线卡住窗口永远不触发。解决办法就是设置空闲分区超时超过这个时间没数据的分区就被标记为空闲不参与水位线计算。这个参数不设作业看起来在跑但就是不出结果新手很容易懵。2.3 文件系统读取批流一体的典型场景从文件系统读数据是批处理场景里最常见的需求也是Flink批流一体能力体现得比较明显的地方。Flink读文件有两种模式一种是流模式文件目录被持续监控新文件来了就处理另一种是批模式一次性把目录下所有文件读完就结束。流模式读文件的关键在于文件监控机制。Flink会定期扫描目录发现新文件就读取。这里有个细节文件必须原子性地出现在目录里否则Flink可能读到一个写了一半的文件。常见的做法是先把文件写到临时目录写完再rename到目标目录rename操作在大多数文件系统上是原子的。批模式读文件则要注意文件分片。Flink会把大文件切成多个分片分给不同的并行子任务去读。分片大小是可以配置的默认值在大多数场景下够用但如果你的文件特别大或者特别小可能需要调整。文件太小会导致分片过多、调度开销大文件太大则单个分片处理时间长影响并行度利用率。// 流模式读文件目录 DataStreamString stream env.readFile( new TextInputFormat(new Path(/data/input)), /data/input, FileProcessingMode.PROCESS_CONTINUOUSLY, 60000L // 每60秒扫描一次新文件 );上面这段代码里PROCESS_CONTINUOUSLY表示持续监控目录60000L是扫描间隔。如果你只想读一次就结束把模式改成PROCESS_ONCE即可。扫描间隔的设置需要权衡设太短频繁扫描目录增加NameNode压力设太长新文件发现延迟高。我一般根据数据到达频率来定数据来得快就设短一点比如10秒来得慢就设长一点比如5分钟。2.4 集合与Socket读取只适合本地调试fromCollection、fromElements、socketTextStream这几个方法我强烈建议只在本地调试时用千万别带到生产环境。原因很简单集合类Source的数据全在内存里数据量一大就OOMSocket Source则是单并行度的而且没有容错保证作业挂了数据就丢了。我见过有人图省事用fromCollection读一个几万条的配置数据本地跑没问题一上生产数据量涨到几百万条直接内存溢出。这类Source的正确用法是本地写单元测试、验证算子逻辑时用一下验证完就换成正经的Connector。选型这件事说到底就一句话生产环境优先用官方Connector特殊来源用自定义SourceFunction集合和Socket只用于调试。把这句话记住能避开一大半选型上的坑。3. 并行度、分区与反压数据读取性能的三根支柱选好了读取方式接下来要解决的是读得快不快、稳不稳的问题。这一章我围绕并行度、分区策略、反压这三个核心概念展开它们共同决定了数据读取环节的性能表现。3.1 并行度设置不是越大越好并行度是Flink里最基础也最容易被误解的概念。很多人觉得并行度开大点总没错机器多嘛多开点并行度跑得快。但实际上Source端的并行度设置有一套自己的逻辑盲目开大反而会坏事。对于Connector类的Source并行度通常和数据源的分区数挂钩。比如读一个消息队列如果队列有16个分区那Source并行度最多设成16再大也没用多出来的SubTask分不到数据纯属浪费资源。反过来如果并行度小于分区数那一个SubTask要处理多个分区单SubTask压力大整体吞吐上不去。所以Source并行度的黄金法则是并行度等于数据源分区数。这样每个SubTask正好处理一个分区负载最均衡。如果数据源分区数会动态变化那就把并行度设成分区数的上限配合分区发现机制新分区出现时自动分配给空闲的SubTask。对于自定义SourceFunction并行度设置更灵活但也更需要想清楚。如果你的Source是从一个单点的HTTP接口拉数据那并行度设成1就够了设多了反而会重复拉取。如果是从多个数据源分片拉取那并行度可以等于分片数。这里有个容易被忽略的点Source并行度和下游算子并行度是独立的。你可以Source开4个并行度下游map开8个Flink会自动做数据重分布。但重分布是有代价的跨节点的数据传输会走网络增加延迟。所以如果Source和下游并行度不一致最好在Source后面接一个rebalance()或者rescale()显式控制数据怎么分发而不是让Flink默认的转发策略来决定。3.2 分区策略数据怎么分决定了后面怎么算分区策略解决的是一条数据该去哪个SubTask的问题。Flink内置了好几种分区器每种适用的场景不一样。forward是默认策略数据在同一个SubTask里流转不跨节点。这种策略效率最高但要求上下游并行度一致否则会报错。rebalance是轮询分发把数据均匀地打散到所有下游SubTask适合数据倾斜的场景。rescale是本地轮询只在同一个TaskManager内部轮询减少网络传输。keyBy则是按key的哈希值分区保证相同key的数据去同一个SubTask是做聚合计算的前提。我重点说一下keyBy因为它是数据读取之后最常用的分区方式也是最容易出问题的。keyBy的坑在于key的哈希分布可能不均匀。如果你的key是用户ID而某些大V用户的活跃度是普通用户的几千倍那这些大V的key就会集中到少数几个SubTask上造成严重的数据倾斜。解决办法是在key里加盐比如把userId改成userId _ random(0, 10)先打散再聚合聚合完再去盐。这个技巧在数据读取阶段就要想好因为keyBy通常紧跟在Source后面。还有一个细节keyBy之后的数据流是KeyedStream只能做keyed相关的操作。如果你后面想做一个全局的聚合就得先把KeyedStream转回DataStream用windowAll或者process来做。这个转换在数据读取阶段就要规划好不然写到一半发现算子用不了返工成本很高。3.3 反压数据读取环节最该警惕的信号反压是Flink里一个非常重要的概念它描述的是下游处理不过来压力向上游传导的现象。反压本身不是bug它是Flink的一种自我保护机制。但如果反压长期存在说明系统里有瓶颈必须处理。反压的根源通常有三个一是数据倾斜某些SubTask处理的数据量远大于其他SubTask二是算子逻辑太重比如在map里做了耗时的外部调用三是资源不足CPU、内存、网络带宽不够。排查反压我一般按这个顺序来先看Flink Web UI的BackPressure页面确认是哪个算子出现了反压然后看该算子的各SubTask处理速率如果差异很大基本就是数据倾斜如果差异不大但整体速率低那就是算子逻辑或资源的问题。数据读取环节的反压往往表现为Source的发送速率被下游限制。这时候你要判断是Source发得太快还是下游处理得太慢如果是Source发得太快可以给Source加一个限流比如用RateLimiter控制每秒发送的数据量。如果是下游太慢那就得优化下游算子的逻辑或者增加下游的并行度。注意反压是症状不是病因。看到反压不要急着调Source的参数先定位到真正的瓶颈算子对症下药才有用。4. 水位线、Checkpoint与数据一致性读取环节的隐藏考点数据读取不只是把数据读进来这么简单它还牵扯到水位线生成、Checkpoint对齐、数据一致性这些深层次的问题。这一章我把这些隐藏考点挖出来讲清楚。4.1 水位线在Source端是怎么生成的水位线是事件时间语义的核心它标记了事件时间推进到了哪个点。水位线可以在Source端生成也可以在下游算子生成但最推荐的做法是在Source端生成因为Source最清楚数据的真实时间分布。在Source端生成水位线有两种常见方式周期性生成和打点生成。周期性生成是每隔一段时间比如200毫秒根据当前看到的最大事件时间减去一个延迟生成一个水位线。打点生成则是在遇到特定标记比如某个特殊事件时生成水位线。周期性生成更常用因为它不依赖数据里有没有特殊标记。延迟的设置是关键延迟太小迟到数据会被丢弃延迟太大窗口触发慢结果出得晚。我的经验是先统计一下数据的乱序分布取P99的乱序时长作为延迟再乘以1.5作为安全系数。WatermarkStrategyString strategy WatermarkStrategy .StringforBoundedOutOfOrderness(Duration.ofSeconds(5)) .withTimestampAssigner((event, timestamp) - extractTimestamp(event)) .withIdleness(Duration.ofMinutes(1));上面这段代码里forBoundedOutOfOrderness(Duration.ofSeconds(5))表示允许5秒的乱序withIdleness(Duration.ofMinutes(1))就是前面提到的空闲分区超时。这两个参数配合使用能解决大部分水位线卡住的问题。4.2 Checkpoint对齐为什么你的作业恢复后数据对不上Checkpoint是Flink容错的核心机制它定期把作业的状态快照保存下来作业失败时从最近的快照恢复。数据读取环节的Checkpoint核心要解决的是偏移量和数据的一致性问题。以消息队列为例Source需要记录我读到哪个偏移量了。做Checkpoint的时候Flink会把当前偏移量保存下来。恢复的时候从保存的偏移量继续读。听起来很简单但如果偏移量的保存和数据的发送不是原子的就会出问题。具体来说如果Source先更新了偏移量、再发送数据那Checkpoint保存的偏移量可能比实际发送的数据超前恢复时会丢数据。反过来如果先发送数据、再更新偏移量恢复时会重复读数据。Flink的解决方案是用CheckpointLock把这两个操作锁在一起保证原子性。这也是为什么前面自定义SourceFunction的例子里collect要包在synchronized块里。对于Connector类的Source这些逻辑Connector已经帮你处理好了你只需要确保Checkpoint是开启的、间隔设置合理就行。Checkpoint间隔太短频繁做快照影响性能间隔太长恢复时需要重放的数据多恢复时间长。我一般设成1到5分钟具体看数据量和恢复时间要求。4.3 端到端一致性读取只是其中一环端到端一致性指的是从Source读进来到Sink写出去整个过程数据不丢不重。这是一个系统工程数据读取只是其中一环但这一环没做好后面再完美也白搭。实现端到端一致性Source端需要支持可重放。也就是说作业失败恢复后Source能从上次的偏移量重新读数据。消息队列类Source天然支持这一点文件类Source通过记录已读文件列表也能做到但Socket类Source就不行——数据读走就没了恢复不了。所以如果你对一致性有要求选Source的时候就要把可重放作为硬性条件。我见过有人用Socket Source做生产作业还开了Checkpoint以为万无一失结果作业一挂数据就丢了Checkpoint根本救不回来。这个坑踩一次就够记一辈子。5. 那些年我在数据读取上踩过的坑前面讲的都是应该怎么做这一章我换个角度讲讲我实际踩过哪些坑。这些坑有的是配置问题有的是理解偏差有的是版本差异每一个都让我花了不小代价才搞明白。5.1 起始偏移量设成最新结果历史数据全丢了这是我早期犯的一个典型错误。当时接了一个需求要把消息队列里的数据实时同步到下游。我配置Source的时候看到有个起始偏移量选项心想实时同步嘛肯定从最新数据开始读就选了最新。作业跑起来看着挺正常数据也在往下游写。结果第二天业务方找过来说昨天的数据怎么少了一大截。一查才发现我选最新意味着作业启动那一刻之前的消息全都不读了而作业启动前队列里积压了大量数据全被跳过了。正确的做法是选最早把积压的数据先消费掉再进入实时模式。或者更稳妥的做法是先记录当前偏移量作业从记录的位置开始读。这个坑的教训是起始偏移量的选择取决于你的业务语义。如果是纯实时、不关心历史选最新没问题如果要做数据同步、不能丢数据必须选最早或者指定偏移量。选之前一定要跟业务方确认清楚。5.2 并行度改了但数据分布没变有一次优化作业性能我把Source的并行度从2调到了8满心期待吞吐量能翻几倍。结果跑起来一看吞吐量几乎没变8个SubTask里有6个是空闲的数据全挤在2个SubTask上。排查了半天才明白我用的Source是自定义的内部按某个字段做了分片但分片逻辑写死了只分2片。并行度调到8多出来的6个SubTask分不到任何分片自然就空闲了。解决办法是改分片逻辑让它能根据并行度动态分片。这件事让我明白一个道理并行度不是孤立的参数它必须和数据源的分片能力匹配。Source能分多少片并行度才能开多大。改并行度之前先确认Source支不支持这么多分片。5.3 水位线延迟设太小迟到数据全被丢做实时统计的时候我给水位线设了1秒的延迟想着数据应该不会乱序太多。结果上线后发现统计结果总是比实际少一点。查了数据才发现有大约3%的数据乱序超过了1秒这些数据到达时窗口已经触发过了直接被丢弃。后来我把延迟调到了5秒数据对上了但窗口触发变慢了业务方又抱怨结果出得晚。最后折中方案是延迟设3秒同时开启允许迟到让迟到数据能更新已经触发过的窗口结果。这样既保证了数据准确又不会让结果等太久。这个坑的教训是水位线延迟是一个权衡没有完美值。你要在数据准确性和结果实时性之间找平衡点。我的经验是先统计乱序分布取P95或P99作为延迟再配合允许迟到机制兜底。5.4 版本升级后Source的API变了Flink的API演进比较快不同版本之间Source的写法可能有差异。我有一次把一个作业从旧版本升级到新版本发现原来的addSource方法被标记为过时了推荐用新的fromSource。虽然旧方法还能用但编译时会有一堆警告而且新版本的一些特性比如新的水位线策略在旧API上用不了。升级的时候我建议先把Source相关的代码单独拎出来对照官方迁移文档改一遍改完在测试环境充分验证再上生产。别小看API变更有时候一个方法签名的变化背后是整套机制的调整不仔细看文档很容易改错。5.5 忘了设空闲分区超时窗口永远不触发这个坑前面提过但因为它太典型了我想再展开说一下。当时的情况是数据源有10个分区白天10个分区都有数据作业跑得好好的。到了晚上只有3个分区有数据另外7个空了。结果从晚上开始窗口就不触发了下游一直收不到结果。排查的时候我先看了数据确认数据在正常进来又看了窗口逻辑确认没问题最后看水位线发现水位线卡在几个小时前不动了。这才想到是空闲分区的问题——7个空分区的水位线不推进拖累了整个作业的水位线。加上withIdleness配置后问题解决。这个坑的隐蔽性在于它只在特定条件下触发部分分区空闲平时测试根本发现不了。所以我现在养成了一个习惯只要是多分区的Source一律加上空闲分区超时配置宁可多设一个参数也不要留一个定时炸弹。6. 从读取到计算如何设计一条顺畅的数据链路数据读取不是终点它是一条数据链路的第一环。这一章我想跳出Source本身聊聊怎么把数据读取和后续的计算环节串起来设计一条顺畅的链路。6.1 Source之后的第一件事想清楚数据怎么分Source读进来的数据第一件事就是决定怎么分给下游。这个决策影响深远因为一旦分了后面的算子都得按这个分法来。如果下游要做key级别的聚合那Source之后立刻keyBy让相同key的数据尽早汇聚到一起减少后续shuffle。如果下游要做全局聚合那Source之后用rebalance打散让所有SubTask负载均衡。如果下游只是简单的map过滤那用forward保持本地性效率最高。我的经验是在Source之后加一个显式的分区算子不要依赖默认策略。默认策略虽然省事但它的行为不一定符合你的预期出了问题也不好排查。显式写出来代码即文档后面维护的人一看就懂。6.2 数据清洗放在哪Source里还是Source后数据清洗比如过滤脏数据、格式转换放在哪是个值得讨论的问题。放在Source里可以减少下游的数据量但会让Source的逻辑变重放在Source后单独一个算子逻辑清晰但多了一次数据传递。我的建议是轻量清洗放Source后重量清洗放Source里。轻量清洗比如过滤null值、简单的字段提取放在Source后的filter和map里逻辑清晰也方便单独测试。重量清洗比如复杂的正则匹配、外部接口调用放在Source里避免这些耗时操作影响下游算子的并行度。还有一个考虑是错误处理。如果清洗过程中可能出错放在Source后更容易做旁路输出把错误数据引到单独的流里。放在Source里的话错误处理逻辑会和读取逻辑混在一起不好维护。6.3 监控与告警数据读取环节该盯哪些指标作业上线不是终点还得盯着它跑。数据读取环节有几个关键指标我建议都配上监控和告警。第一个是消费延迟也就是最新数据的时间和当前处理时间之间的差距。这个指标直接反映作业跟不跟得上数据产生的速度。延迟持续增长说明处理能力不足得扩容或者优化。第二个是Source的吞吐量也就是每秒读进来多少条数据。这个指标突然下降可能是数据源出问题了也可能是作业内部有反压。第三个是Checkpoint的成功率和耗时。Checkpoint频繁失败或者耗时过长说明状态太大或者资源不足需要调整。第四个是水位线推进情况。水位线长时间不推进要么是没数据要么是空闲分区没处理得赶紧排查。这些指标在Flink Web UI上都能看到但生产环境一般会接到监控系统里配上阈值告警。我的习惯是消费延迟超过5分钟告警Checkpoint连续失败3次告警水位线超过10分钟不推进告警。阈值可以根据业务特点调整但一定要有告警不能等业务方找上门才发现问题。7. 一些关于数据读取的个人体会写了这么多最后我想分享几点偏个人化的体会不一定对但都是我真金白银换来的。第一点数据读取的复杂度被严重低估了。很多人觉得Source就是个入口随便写写就行把精力都放在计算逻辑上。但实际上一个作业出问题十有八九能追溯到Source端。并行度、分区、水位线、Checkpoint每一个都是深水区。我现在的习惯是接到一个新作业先花一半时间把Source端设计清楚剩下的时间再写计算逻辑。第二点配置参数没有标准答案只有适合当前场景的答案。水位线延迟设多少、Checkpoint间隔设多长、并行度开多大这些都没有放之四海而皆准的值。你得结合自己的数据特点、业务要求、集群资源来定。别人分享的配置可以参考但别照搬一定要在自己的场景里验证过再用。第三点测试环境永远模拟不出生产的全部问题。数据倾斜、空闲分区、反压这些问题在测试环境里往往因为数据量小、场景简单而暴露不出来。所以上线之后要盯紧监控发现问题及时处理。我现在的做法是新作业上线后前三天每天看一次监控一周后改成每周看一次稳定运行一个月后才完全放手。第四点文档和源码是最好的老师。Flink的官方文档写得很详细很多参数的含义和推荐值都有说明。遇到文档讲不清楚的直接看Connector的源码看它怎么处理偏移量、怎么生成水位线、怎么做Checkpoint。源码看多了很多玄学问题就变成常识问题了。数据读取这件事说难不难说简单也不简单。它不需要你懂多高深的算法但需要你对Flink的运行机制有扎实的理解对配置参数的影响有清晰的认知对可能出现的坑有足够的警惕。把这一层打扎实了后面的计算、状态、容错学起来都会顺很多。希望这篇内容能帮你少走一些弯路把数据读取这一关稳稳地过掉。