1. 依赖准备：flink-connector-files
Java 工程要使用文本文件 Source，需要引入 Flink 的文件连接器依赖：
<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-files</artifactId><version>2.2.0</version></dependency>PyFlink 用户通常可以直接在作业里使用（但如果你集群环境缺少对应 jar，也需要通过 Python dependency management 方式携带）。
2. 为什么用 FileSource + TextLineInputFormat
TextLineInputFormat 解决两件事：
- 按行切分：每行一个 record
- 字符集解码：使用
InputStreamReader支持多种 charset（UTF-8、GBK 等）
而 FileSource 解决两件事：
- bounded：一次性读完目录/文件（批处理）
- continuous：持续监控目录，新文件出现就继续读（流式文件输入）
也就是说，你可以用同一套 Source，覆盖两类场景：
- 离线回放历史日志
- 实时消费不断落盘的新日志文件
3. 批处理模式（Bounded）：读完就结束
目标：把一个文本文件（或目录里所有文本文件）的每一行读成String，生成DataStream<String>。
因为文本行一般不自带事件时间，所以不需要 watermark：
finalFileSource<String>source=FileSource.forRecordStreamFormat(newTextLineInputFormat(),/* Flink Path */).build();finalDataStream<String>stream=env.fromSource(source,WatermarkStrategy.noWatermarks(),"file-source");适用场景：
- 跑一次把历史文件处理完（ETL、离线修数、回放）
4. 流处理模式（Continuous）：持续监控目录，新文件不断加入
目标：目录持续落文件（例如按小时切日志），Flink 任务一直跑，新文件出现就读，DataStream 会“无限增长”。
通过monitorContinuously(Duration)开启目录监控，比如每 1 秒扫描一次：
finalFileSource<String>source=FileSource.forRecordStreamFormat(newTextLineInputFormat(),/* Flink Path */).monitorContinuously(Duration.ofSeconds(1L)).build();finalDataStream<String>stream=env.fromSource(source,WatermarkStrategy.noWatermarks(),"file-source");适用场景：
- 应用日志落盘目录（log rolling）
- 上游系统定时导出文件到目录
- 简易的“文件流”采集管道（没有 Kafka 也能跑）
5. 生产建议：文本文件“流式监控”最容易踩的坑
5.1 只监控“新文件”，不等于“追尾追加写”
大多数文件监控模式更适合“文件落地后不再变”（写完再 rename/commit）。如果你希望读一个不断追加的单文件（类似tail -f），要非常谨慎：有些文件系统/写入方式会导致重复读或读到半行。
推荐的落地方式：
- 上游写临时文件（
.tmp），写完后rename 成正式文件名 - Flink 只消费正式文件名规则（例如不匹配
.tmp）
5.2 监控频率不是越小越好
monitorContinuously(1s)会频繁扫描目录：
- 目录文件数大时会产生明显压力
- 对对象存储（S3/OSS）类系统，list 成本更高
经验：
- 本地/小目录：1s～5s 可以
- 大目录/对象存储：10s～60s 起步，并控制目录分区层级（按日期/小时分层）
5.3 字符集与脏数据治理要提前考虑
TextLineInputFormat 基于InputStreamReader解码，编码不一致会出现乱码或异常。建议：
- 统一上游编码（最好 UTF-8）
- 对异常行做侧输出（side output）或打到 DLQ（如果你后续接 Kafka）
5.4 文本行没有事件时间时，watermark 怎么办
如果你的行里其实包含时间戳（比如日志行开头有2026-01-15 12:34:56），你可以在 map/flatMap 里解析事件时间，再配置 watermark 策略；否则默认 noWatermarks 没问题。
6. 一句话总结
- TextLineInputFormat：把文件按“行”读成
String，并处理字符集解码 - FileSource：同一套代码支持批（bounded）与流（continuous 目录监控）
- 批：
.build()直接读完结束 - 流：
.monitorContinuously(Duration)目录新文件持续进入
)
