企业官网建设流程全解析

上周有个做离线报表的朋友跑来问，说他们想把落盘的一批日志文件接进 Flink，顺手写了句env.readTextFile，结果 IDE 里整行都是灰的，翻文档换成env.fromSource又不知道参数该怎么填。这事儿在 Flink DataSource 这块特别典型：同样是“读文件”，框架前后给了三条完全不同的路，三条路的执行模型、状态管理、断点续传能力差得非常远，选错了轻则作业跑完不退出，重则上线后天天重复消费。这篇就拿“Flink 集成 DataSource 读取文件数据”这一个点，把三条路各自适合什么场景、代码怎么写、坑埋在哪里，一条条拆开讲清楚。

先说清楚这篇适合谁看：如果你正在用 Flink 做文件摄取，或者被SourceFunction is deprecated的警告困扰，再或者你写了个文件源发现 checkpoint 越做越大、作业重启后数据翻倍，那这篇基本能对上号。我会从旧的SourceFunction手写实现一路讲到 FLIP-27 的FileSource，中间夹着实测踩出来的问题排查链路，代码都能直接抄。

1. 文件数据源在 Flink 里到底卡在哪

很多人下意识觉得“读文件”是最简单的一类数据源，比起 Kafka、消息队列，文件又没有分区再均衡、又没有消费位点，能难到哪去。真正上手才发现恰恰相反：文件源是 Flink 里概念最拧巴的一类源，因为它同时横跨了批处理和流处理两种世界观。批处理视角下文件是有界的、可以整体切分并行读的；流处理视角下文件目录是一个持续变化的东西，新文件不断落进来，源要一直醒着。Flink 早期用一套FileInputFormat硬扛两种语义，后来发现扛不住，才有了 FLIP-27 整个 Source API 的重构。理解这一点，后面的所有 API 选择就有了依据。

1.1 批读与流读的分歧点：谁来决定“读完了”

分歧的核心只有一个问题——源什么时候算结束。有界源读完所有输入后，会发出一个代表“不会再有数据”的信号，下游算子收到之后触发窗口关闭、作业进入 FINISHED 状态；无界源永远不发这个信号，它就那么挂着等新数据。文件源尴尬的地方在于，同一份文件目录，你既可以当成“一次性全量导入”（有界），也可以当成“持续监控增量落盘”（无界），两种用法在业务上都极其常见。

这个选择在代码里对应的是Boundedness这个枚举，取值就两个：BOUNDED和CONTINUOUS_UNBOUNDED。听起来简单，但实际踩坑率极高。最典型的错误是把一个本该有界的离线导入写成了持续监控模式，作业跑完之后一直不结束，调度系统以为任务卡死，运维半夜被叫起来看监控。反过来把流式摄取写成了有界模式，作业启动那一刻目录里只有两个文件，读完就退出，后面落进来的几百个文件全被忽略，数据静默丢失，这种问题比前面那种更阴险，因为作业“看起来跑成功了”。

所以第一步永远不是写代码，是先想清楚：这个文件目录会不会在作业运行期间持续有新文件进来？会，就是无界；不会，就是有界。这个答案决定了后面所有的 API 选型和配置。

1.2 三个入口的能力边界对比

Flink 读文件目前存在三个入口，很多人混淆就是因为不清楚它们各自的定位。下面这张表是我按实际使用体验整理的，比官方文档更贴近真实项目的取舍：

四个入口里，前三个都已经是历史包袱。readTextFile最省事但也最弱，它连并行切分都做不好，只适合写 demo 和本地调试；readFile曾经是文件流式摄取的主力，靠FileProcessingMode区分一次性读和持续监控，但它把有界性和监控逻辑糅在一起，状态语义含糊，遇到并行度变化就容易出问题；自定义SourceFunction灵活度最高，什么奇怪的文件命名、什么特殊的分片规则都能塞进去，但代价是状态、位点、容错全得自己扛。

生产环境我现在的默认选择一律是FileSource。它把“文件怎么切分成 split”“split 怎么分配给 reader”“读进度怎么进 checkpoint”这三件事拆得干干净净，每一件都有明确的扩展点。代价是有学习曲线，得先搞明白 SplitEnumerator、SourceReader 这几个角色是干什么的，这部分我在第 2 章展开。

2. Source 接口的换代：从 SourceFunction 到 FLIP-27

Flink 1.12 引入 FLIP-27 的时候，社区里普遍的反应是“又一个抽象层”，觉得老接口能跑就行。两年之后再回头看，这次换代其实是必须的。老SourceFunction的致命问题在于它把两件不同粒度的事绑在了一个run()方法里：一是“我要读哪些数据”（数据发现），二是“我怎么把数据一条条吐出去”（数据读取）。在单并行度、数据源稳定的场景下没问题；一旦涉及到分片发现、动态扩缩容、位点快照，这两件事搅在一起就必然打架。

2.1 SourceFunction 时代文件读取的执行链路

老接口的执行链路非常直白，每个并行子任务起一个线程跑run()方法，线程里自己开文件流、自己读、自己调ctx.collect()。听起来没什么问题，但有几个隐含假设很要命。

第一个假设是“分片逻辑可以在启动时一次性确定”。你的open()方法里数一下目录里有几个文件，按 subtask index 取模分配，这在一个静态目录下完全正确。可一旦作业从 checkpoint 恢复，而且期间目录里新增了文件、或者并行度从 4 调成了 6，那这个取模结果跟之前就对不上了，有的文件被读两遍，有的文件永远读不到。这时候你唯一的补救办法是把“哪个子任务读了哪个文件的哪个位置”这个映射也存进状态里，然后用 union state 在恢复时重新分配——工作量一下子就上去了。

第二个假设是“读取位点只有我自己关心”。SourceFunction本身不提供任何位点管理，你要做到精确一次，就得在snapshotState里把所有打开文件的 offset 序列化进去，在initializeState里读回来并 seek。文件句柄这种不可序列化的东西还得靠transient加手动重开。这套活儿写下来两百行打底，还得自己测各种异常恢复场景。

所以老接口不是不能用，是“写对了的成本很高，写错了也看不出来”。很多线上重复消费的问题，根子都在这儿。

2.2 新 Source API 的三件套拆解

FLIP-27 把源拆成了三个角色，每个角色职责单一：

Split是最小可分配的数据单元。对文件源来说，一个 split 通常对应一个文件，或者大文件切出来的一段字节区间。Split 本身必须可序列化，因为它要被分发到不同节点、要被写进 checkpoint。

SplitEnumerator跑在 JobManager 侧，负责发现有哪些 split、决定分给哪个 reader、记录哪些已经分配、哪些已完成。它自己的状态也会进 checkpoint，这就解决了老接口里“分片映射没法恢复”的问题。

SourceReader跑在 TaskManager 侧，从 enumerator 那里领到 split 之后负责真正读数据、往下游发记录、汇报进度。

这套拆法带来的直接好处是：并行度变化时，enumerator 重新分配未完成的 split 就行，不需要人肉维护映射表；每个 reader 只关心自己手上的 split，进度汇报是增量式的；状态大小跟“正在读的 split 数”相关，而不是跟“总数据量”相关。理解这三件套，后面看FileSource的各种 builder 方法就是水到渠成的事。

2.3 版本对照与依赖坐标

动手之前先对版本。FileSource位于独立的flink-connector-files模块，不在核心包里，这个依赖不加是编译不过的：

<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-files</artifactId> <version>${flink.version}</version> </dependency>

版本差异主要体现在构建方法上，这块不同版本差异很大，抄代码前先核对：

1.14 是分水岭。forRecordStreamFormat这套设计把“格式解析”也抽象掉了，TextLineInputFormat、CsvInputFormat都实现了StreamFormat接口，配上FileSource之后整条链路是统一的。我下面所有示例都以 1.14 之后的写法为准，老版本的项目如果短时间升不上去，思路一样，只是 builder 换成forBulkFileFormat并传一个老的FileInputFormat实例。

顺带提一句SourceFunction的弃用状态：接口从 1.14 起标记为@Deprecated，官方口径是“不保证移除时间，但新代码不要用”。存量代码不急着改，新项目再往里跳就没必要了。

3. 手写 RichSourceFunction 读文件的完整落地

虽然新 API 是首选，但我还是建议每个做文件摄取的人都亲手写一遍RichSourceFunction。原因很实际：一是存量项目里这套代码遍地都是，你得看得懂；二是写完一遍你才会真正明白新 API 帮你省掉了什么，遇到状态异常时才知道往哪查。这一章我给一个能跑的最小实现，然后逐层加容错。

3.1 最小可运行实现与逐行解释

下面这个实现做三件事：把目录下的普通文件按 subtask 索引取模分配、逐行读取、在cancel时能正常退出。

public class LocalFileSource extends RichSourceFunction<String> { private final String dirPath; private volatile boolean running = true; private transient List<Path> assigned; private transient BufferedReader reader; public LocalFileSource(String dirPath) { this.dirPath = dirPath; } @Override public void open(Configuration parameters) throws Exception { int subTask = getRuntimeContext().getIndexOfThisSubtask(); int parallelism = getRuntimeContext().getNumberOfParallelSubtasks(); List<Path> all; try (Stream<Path> s = Files.list(Paths.get(dirPath))) { all = s.filter(Files::isRegularFile) .filter(p -> !p.getFileName().toString().startsWith("_")) .filter(p -> !p.getFileName().toString().startsWith(".")) .sorted() .collect(Collectors.toList()); } assigned = new ArrayList<>(); for (int i = 0; i < all.size(); i++) { if (i % parallelism == subTask) { assigned.add(all.get(i)); } } LOG.info("subtask {} got {} files out of {}", subTask, assigned.size(), all.size()); } @Override public void run(SourceContext<String> ctx) throws Exception { for (Path p : assigned) { if (!running) break; try (BufferedReader br = Files.newBufferedReader(p, StandardCharsets.UTF_8)) { reader = br; String line; while (running && (line = br.readLine()) != null) { synchronized (ctx.getCheckpointLock()) { ctx.collect(line); } } } } } @Override public void cancel() { running = false; try { if (reader != null) reader.close(); } catch (IOException ignored) { } } }

几个点必须解释清楚，否则这段代码很容易被改坏。

open()里做文件发现而不是构造函数里做，是因为构造函数在客户端执行，那时候目录可能还不存在或者权限不同，而open()是在 TaskManager 上执行的，环境更接近真实读取场景。

ctx.getCheckpointLock()这个同步块不是可选项。它保证“发出一条记录”和“记录位点”这两步之间不会被 checkpoint 快照打断。如果不用这把锁，极端情况下你会发出去了数据但位点没记上，恢复后重复读；或者位点记上了但数据还没发出去，恢复后丢数据。加锁会有一点性能开销，但这个代价必须付。

那两个过滤条件是血泪教训：Flink 的文件源默认会忽略以下划线或点开头的文件，很多写入框架会用_SUCCESS这类文件做完成标记，用.tmp做临时文件，你如果不过滤，读到的就是半截数据和一堆空文件。

3.2 并行切分与 open/close 生命周期

取模分配这个策略在小规模场景够用，但它有两个硬伤得知道。文件大小不均的时候，取模分配会导致严重的倾斜——某个子任务分到三个 G 的文件，隔壁分到三个 K 的，整体吞吐就卡在最慢的那个上。真正要均衡，得按字节大小排序后再分配，或者干脆按字节区间切分大文件。FileSource内部做的就是后者，它支持splitMaxSize参数控制单个 split 的字节上限，大于这个值的文件会被切成多段，这才真正解决了倾斜。

生命周期这块，open()和close()一定成对写。我见过因为close()里没关流导致文件句柄泄漏的案例，作业跑一周之后 TaskManager 报Too many open files，整个节点上所有作业一起挂。RichSourceFunction提供了close()钩子，把BufferedReader关掉即可。

另外cancel()里的running = false是volatile的，这个修饰符不能省。取消信号来自另一个线程，不加volatileJVM 不保证可见性，run()里的循环可能一直读不到新值，作业取消失败，最后只能 kill 进程。

3.3 加上 CheckpointedFunction 做到断点续读

上面那版重启就从头读，适用于可重跑的场景。要做到断点续读，得实现CheckpointedFunction，把文件名到 offset 的映射存进算子状态：

public class ResumableFileSource extends RichSourceFunction<String> implements CheckpointedFunction { private final String dirPath; private final Map<String, Long> offsets = new HashMap<>(); private transient ListState<String> offsetState; @Override public void snapshotState(FunctionSnapshotContext context) throws Exception { offsetState.clear(); for (Map.Entry<String, Long> e : offsets.entrySet()) { offsetState.add(e.getKey() + "|" + e.getValue()); } } @Override public void initializeState(FunctionInitializationContext context) throws Exception { ListStateDescriptor<String> desc = new ListStateDescriptor<>("file-offsets", String.class); offsetState = context.getOperatorStateStore().getUnionListState(desc); if (context.isRestored()) { for (String s : offsetState.get()) { int idx = s.lastIndexOf('|'); offsets.put(s.substring(0, idx), Long.parseLong(s.substring(idx + 1))); } } } }

关键在于用的是getUnionListState而不是getListState。区别在于：getListState在恢复时会保持原有的“状态到子任务”的映射关系，并行度一变就对不上了；getUnionListState会把所有子任务的状态合并成一个大列表给每个子任务，然后由你自己重新分配。文件源这种“状态条目和数据本身解耦”的场景，必须用 union 版本。

还有个细节：记录 offset 的时候要按字节 offset 而不是行号。行号看起来更直观，但你恢复时还是得从头扫到那一行，等于没省；字节 offset 可以直接FileChannel.position(n)跳过去。BufferedReader拿不到字节 offset，真要精确控制得换成FileChannel加手动按行切分，这部分代码量不小，也是为什么我最终都切到了FileSource——这些脏活它都干完了。

4. 用 FileSource 走生产级路线

到这一章才算进入正题。FileSource用起来很短，但每个 builder 方法背后都有语义，填错了不会报错，只会行为异常。我把实际项目里最常改的三处单独拎出来讲。

4.1 Boundedness 选错就会挂住的那些场景

FileSource的有界性由构建方式决定。只给路径不给监控参数，它就是一个有界源，读完目录里现有的文件就发结束信号退出；加了monitorContinuously，它就是无界的，一直轮询目录。这两者之间没有“手动切换”的开关，选错了只能改代码重新提。

// 有界：一次性全量导入 FileSource<String> bounded = FileSource .forRecordStreamFormat(new TextLineInputFormat(), new Path("file:///data/inbox")) .build(); // 无界：持续监控增量文件 FileSource<String> unbounded = FileSource .forRecordStreamFormat(new TextLineInputFormat(), new Path("file:///data/inbox")) .monitorContinuously(Duration.ofSeconds(30)) .build();

monitorContinuously的参数是轮询间隔。这个值不要设得太小，我见过设成 1 秒的，几百个监控源同时跑，TaskManager 上全是列目录的 IO，真正的读取任务反而被挤。实践中 30 秒到 5 分钟都是合理区间，取决于你的上游落盘频率。如果是分钟级批产出的上游，30 秒够用；如果是整点批量出的，5 分钟也没问题，晚几分钟触发换来的是稳定的 IO 压力。

env.fromSource的第二个参数是水印策略。文件源通常没有事件时间语义，用WatermarkStrategy.noWatermarks()就行。但如果你确实要从文件内容里抽事件时间，就得写一个正规的WatermarkStrategy，注意配合withIdleness，否则某个 subtask 长时间分不到文件会导致全局水印不前。

DataStream<String> lines = env.fromSource( unbounded, WatermarkStrategy.noWatermarks(), "file-source");

第三个参数是算子名，别省。省了之后在 Web UI 上看到的是默认名字，排查问题时在一排算子中找不出哪个是文件源，尤其当你有好几个文件源的时候。

4.2 目录监控与增量摄取

monitorContinuously所谓的“监控”，本质上就是定期列目录，跟已有文件列表做差集，新出现的文件生成新 split 分给 reader。这里有三个行为需要提前知道。

第一，它只认“目录”不认“单文件”。传一个具体文件路径进去能启动，但后续不会有任何新 split，行为上跟有界源没区别。要增量就得传目录。

第二，文件的可见性判断依据的是“文件是否出现在目录列表里”。这意味着如果你的上游是“先创建一个同名的 tmp 文件，边写边改名”或者“先写完再 mv 过来”的原子写法，那没问题，文件一出现就是完整的。但如果上游是“直接在目标目录里 fopen 然后写”，那 Flink 可能读到一个正在写入的文件，读到半截就以为文件结束了。这种情况必须在源侧做配合：要么上游改成原子改名，要么加setFileFilter过滤掉未完成的文件。这个坑非常隐蔽，因为数据不是丢，是“少”，日志里也看不出来。

第三，目录列表的排序和去重靠的是一个内部的已完成文件记录，这个记录会随 checkpoint 一起存。所以断点恢复之后不会重复读已经读过的文件。但如果你手动从外部删掉了 checkpoint，或者用了不落状态的存储方式，那重启就是把整个目录再读一遍。

4.3 状态后端与文件源的状态膨胀

文件源的状态主要是两部分：已完成文件的路径集合，以及正在读取的 split 的进度。前者随着监控时间线性增长，后者跟并行度成正比。跑一个持续监控的源，一个月下来完成文件路径可能有几十万条，每条路径几百字节，加起来几十兆，看似不多，但 checkpoint 是要序列化传输的，状态越大 checkpoint 越慢。

我踩过一次典型的：某个文件源监控目录每小时落两千个小文件，跑了三周之后 checkpoint 从 2 秒涨到了 40 秒，然后开始超时失败，作业反复重启。排查下来就是完成文件列表太大。解决方法有两个方向，一是加大 checkpoint 超时和调大状态后端的内存，属于治标；二是从源头减少文件数，让上游合并小文件再落地，这才是治本。小文件问题在流式文件摄取里几乎是必踩的坑，能提前跟上游谈好合并策略，比事后调参数省事得多。

5. 反序列化环节：格式、编码与脏数据

读到数据之后要转成对象，这一步看似琐碎，但线上事故密度很高。文件不像消息队列那样有 schema 注册中心，字段多了少了、编码错了、混进一行乱码，全得自己在解析层处理。

5.1 三种反序列化方式的取舍

按数据格式和复杂度，通常就三条路：

我的经验是：只要数据不是百分之百规整，就别用CsvInputFormat。它的类型转换失败会直接抛异常把作业打挂，而文件场景里脏数据几乎必然存在，挂一次就得回滚重跑。用TextLineInputFormat读成字符串再自己map，解析失败就丢旁路，作业不至于停。

JSON 日志类我一般会包一层，用RichMapFunction加个计数器统计解析失败率，失败率超过阈值再告警。这样既不会一条脏数据打挂整个作业，也不会默默吞掉大量数据。

public class SafeJsonParser extends RichMapFunction<String, Event> { private transient Counter badCounter; @Override public void open(Configuration parameters) { badCounter = getRuntimeContext() .getMetricGroup() .counter("parse_failed_total"); } @Override public Event map(String line) { try { return JSON.parseObject(line, Event.class); } catch (Exception e) { badCounter.inc(); return null; } } }

返回null之后下游接一个filter(Objects::nonNull)就过滤掉了。这个计数器十分关键，它让你能在监控面板上直观看到脏数据比例，而不是靠人肉翻日志。

5.2 编码、换行符、BOM 这些琐碎但致命的事

编码问题在文件场景里的出现频率远高于消息队列，因为消息队列的生产方通常统一，而文件可能来自各种历史系统和脚本。

第一类是UTF-8和GBK混用。TextLineInputFormat默认按 UTF-8 解码，遇到 GBK 文件里的中文，解出来是乱码但不报错，一路流到下游变成脏数据。如果数据源不可控，稳妥做法是先按字节读进来，做一次编码探测再决定解码方式。探测逻辑不复杂，看几个字节的模式就能区分大部分情况。

第二类是 BOM 头。Windows 上导出的 UTF-8 文件经常带一个不可见的 BOM，它会被当成第一个字段的一部分，导致你的 JSON 解析第一行永远失败。处理方式是在解析前line.replace("\uFEFF", "")剥掉。

第三类是换行符。\r\n和\n混用会让readLine()的边界判断出现偏差，偶尔出现末尾多一个不可见字符导致数字解析失败。这个在跨平台数据交换里极其常见，遇到“明明看着对但就是解析失败”，先 hexdump 看一眼行尾。

5.3 脏数据的兜底与旁路输出

脏数据不能只靠丢弃。丢掉的同时必须留下证据，否则出了数据对不上的问题，你连从哪查都不知道。我的固定做法是两条链路：主链路解析成功的数据往下走，失败的数据连同错误原因、文件名、行号一起发到旁路。

旁路最简单的实现是sideOutput：

final OutputTag<String> deadLetter = new OutputTag<String>("dead-letter") {}; SingleOutputStreamOperator<Event> main = lines .process(new ProcessFunction<String, Event>() { @Override public void processElement(String value, Context ctx, Collector<Event> out) { try { out.collect(JSON.parseObject(value, Event.class)); } catch (Exception e) { ctx.output(deadLetter, value); } } }); DataStream<String> bad = main.getSideOutput(deadLetter); bad.addSink(/* 写入独立的落盘目录或日志系统 */);

旁路的落地位置最好是独立目录，文件名带上时间戳，方便后续回放。这套机制建好之后，脏数据处理就从“事故”变成了“日常运维”，补数据的时候直接把旁路文件修好重新投进源目录就行，不用改代码重跑全量。

6. 实测问题与排查链路

下面这几个问题是我在文件摄取上反复遇到的，每一个都值得写进排查手册。我按“现象—怀疑方向—验证方法—根因”的顺序写，方便你直接照着复现。

6.1 文件读完了但作业不结束

现象最直观：数据都出去了，下游也收到了，但作业状态一直卡在 RUNNING，不进入 FINISHED。

第一个要确认的是源的Boundedness。把monitorContinuously去掉试试，如果作业立刻正常结束，那根因就是无界源，行为符合预期，只是你的预期错了。这种情况在“批任务调度系统 + Flink 无界源”的组合里特别常见，因为调度系统默认任务会自己退出。

第二个怀疑方向是水印。有界源读完会发MAX_WATERMARK，但如果下游有算子依赖水印触发（比如事件时间窗口），而水印策略写成了noWatermarks，那水印永远不到最大值，窗口永远不关，作业自然结束不了。验证方法是看 Web UI 里下游算子的 Watermark 指标是不是一直是Long.MIN_VALUE。是的话，换成正规的水印策略。

第三个方向比较隐蔽：withIdleness没配导致的部分 subtask 空转。并行度 8，但目录里只有 3 个文件，5 个 subtask 分不到数据，它们不发水印，全局水印被拖在最小值。这种“数据都处理完了但窗口不触发”的情况，加上withIdleness(Duration.ofMinutes(1))就能解决。

6.2 并行度大于文件数导致的空跑与倾斜

并行度 10、文件 3 个，这是很常见的配置不匹配。后果有两个层面：一是资源浪费，7 个 slot 白占着；二是如果用了union类型的操作或全局聚合，这 7 个空 subtask 会成为瓶颈。

验证很简单，看 Web UI 上每个 subtask 的numRecordsIn，一堆是 0 就是了。处理方法有三条：让上游把文件合并到大于等于并行度的数量；用splitMaxSize把大文件切细；或者干脆降低源的并行度。我个人偏好第三条，文件源本来就不该成为整个作业的并行度瓶颈，读取之后的计算再开高并行度更合理。

倾斜的排查稍微复杂点。看每个 subtask 的numBytesIn或者读取耗时，如果某个 subtask 处理的数据量是别人的十倍，那就是切分不均。FileSource的splitMaxSize能缓解，但它只对单个大文件生效，如果目录里是小文件扎堆，还是得靠上游合并。

6.3 断点续传后数据重复或丢失

这是最需要小心的一类问题，因为它不会报错，只会让数据对不上。

重复的第一嫌疑是“没开 checkpoint 就开始读文件”。不开 checkpoint 就没有位点，重启只能从头来。第二嫌疑是 state backend 配置问题，比如用了内存后端但作业规模已经超了，checkpoint 实际上没成功，恢复时用的是很早的状态。第三嫌疑是源的uid被改过，改了 uid 之后新作业找不到旧状态，直接当全新作业启动。

验证方法很直接：在源后面接一个map，把每条记录的“文件名 + 字节偏移”打出来。正常重启之后，应该从旧位点后面继续，而不是从某文件开头重来。这个验证手段我在每个新文件源上线前都会跑一次，花五分钟，省后面几小时。

丢失的一般和“文件可见性”有关，也就是 4.2 里说的半写文件问题。还有一个可能是状态里的已完成文件列表在恢复过程中被错误清理，这个得看 checkpoint 的实际内容，用状态检查工具 dump 出来比对。

6.4 用指标、日志和背压定位问题

排查不能靠猜。文件源必须提前打好三个观测点。

第一个是自定义计数器，记录“已发现文件数”“已完成文件数”“解析失败数”。这三个数在 Web UI 上一眼就能看出是不是卡住了：发现数涨但完成数不涨，说明读得慢；完成数涨但下游记录数不涨，说明解析出问题了。

第二个是输入速率。文件源的numRecordsInPerSecond在监控面板上应该是一条平稳的线。如果它周期性地掉到 0 再冲高，那说明批次间隔，可能上游落盘是批量的，也可能monitorContinuously的间隔太长。

第三个是背压。文件读取本身不复杂，慢通常都慢在下游。Web UI 变红的位置就是瓶颈，往上游找就行。我遇到过因为后面接了一个每来一条就用外部客户端查一次外部存储的RichMapFunction，导致整条链路吞吐被压到每秒几十条，表现是“文件读得极慢”，实际是下游堵了。

7. 几条从踩坑里换来的经验

写到这里，最后分享几条纯粹的经验，都是文档里不写、但实际决定项目成败的东西。

第一条，文件源的上游契约比代码重要。你花三天把源写得再健壮，只要上游是“直接在目标目录流式写”的，那半截文件的坑永远堵不严。上线前一定要跟写文件的一方把落地方式谈死：先写临时目录、写完原子改名到目标目录、文件命名带明确的时间戳前缀。这三条谈下来，后面省掉的问题比代码优化多十倍。

第二条，别迷信“自动发现新文件”，给它加个兜底。monitorContinuously轮询目录在绝大多数情况下是可靠的，但遇到网络存储的目录缓存延迟，偶尔会出现文件已经落地但列目录看不到的情况，几十秒后才出现。如果你的业务对时效敏感，可以额外加一条轻量的定时任务，定期把“应该出现但还没读到”的文件清单打出来对账，比出事之后再查好得多。

第三条，小文件在这个场景里是真敌人。不管是状态膨胀、checkpoint 变慢，还是空 subtask 空跑，根源大多能追到小文件上。能推动上游合并就推，推不动就在摄取层加一个预合并的离线环节，把每小时两千个文件合成十个，投入产出比极高。

第四条，上线前那次“杀进程重启”的验证不能省。手动把 TaskManager 杀掉，看作业恢复之后数据是重复了还是缺失了，这一条验证能暴露出九成的状态问题。我见过太多“跑得好好的”作业，直到某次真故障重启才发现吃了三个月重复数据。

这套东西摸下来，最省心的路径其实很清楚：新项目直接上FileSource，有界无界看业务定，格式解析用TextLineInputFormat加自己兜底，旁路一定建，监控一定打。手写SourceFunction那套留在脑子里当知识储备就行，用来读老代码和排查老问题，不必再拿它开新坑。