PyFlink DataStream Word Count 双模式实战：从批处理到流式统计的完整实现-平芜编程栈

大数据
流处理
批处理
数据工程

【免费下载链接】flink

项目地址：https://gitcode.com/gh_mirrors/fli/flink

导读

本文围绕 Flink Python DataStream 示例文档展开，深入剖析 PyFlink 官方提供的两个 Word Count 示例：批处理版 word_count.py 与流式版 streaming_word_count.py。读者将掌握如何使用StreamExecutionEnvironment构建数据流管道，如何通过FileSource/from_collection/datagen三种方式定义数据源，如何用flat_map、map、key_by、reduce完成分词与统计，以及如何用FileSink或print()输出结果。同时结合 PyFlink 源码，说明每个关键 API 的底层行为与适用场景，帮助读者写出可直接运行、可扩展到真实业务的 PyFlink 作业。

示例文档与代码定位

word_count.rst是 PyFlink 官方文档 DataStream 示例索引中的第一篇，通过literalinclude指令直接嵌入两个示例的完整源码，这意味着文档即代码：仓库中 pyflink/examples/datastream 目录下的源文件是文档的唯一权威内容来源。

示例文件	处理模式	数据来源	核心知识点
word_count.py	批处理（BATCH）	内存集合或文本文件	`FileSource`、`RuntimeExecutionMode.BATCH`、`FileSink`
streaming_word_count.py	流处理（STREAMING）	`datagen`连接器持续生成数据	Table 与 DataStream 互转、无限流处理

一、批处理 Word Count：经典分词统计的完整实现

批处理示例的核心代码位于 word_count.py，主函数word_count(input_path, output_path)接受输入与输出路径两个可选参数，完整构建了一条「读取 → 分词 → 计数 → 输出」的数据流管道。

1.1 准备执行环境与运行模式

env = StreamExecutionEnvironment.get_execution_environment() env.set_runtime_mode(RuntimeExecutionMode.BATCH) # write all the data to one file env.set_parallelism(1)

StreamExecutionEnvironment.get_execution_environment()是 PyFlink 所有 DataStream 作业的入口，定义在 stream_execution_environment.py。
RuntimeExecutionMode.BATCH表示以批语义执行：任务全部部署完成后才开始执行，事件时间与处理时间按批语义处理。枚举定义见 execution_mode.py。与之相对的是STREAMING模式——所有任务先部署、开启 checkpoint，完整支持处理时间与事件时间。
set_parallelism(1)将整个作业的并行度设为 1，注释明确说明目的是"把所有数据写入同一个文件"，保证输出结果集中、便于查看。

1.2 三种输入路径：文件读取与内存集合

示例对input_path做了分支处理，演示了两种数据源定义方式：

if input_path is not None: ds = env.from_source( source=FileSource.for_record_stream_format(StreamFormat.text_line_format(), input_path) .process_static_file_set().build(), watermark_strategy=WatermarkStrategy.for_monotonous_timestamps(), source_name="file_source" ) else: print("Executing word_count example with default input data set.") print("Use --input to specify file input.") ds = env.from_collection(word_count_data)

方式一：FileSource读取文本文件。StreamFormat.text_line_format(charset_name="UTF-8")按行读取文件，底层委托给 Java 的TextLineInputFormat，使用java.io.InputStreamReader按指定字符集解码字节流（参见 file_system.py）。FileSource.for_record_stream_format(...)构建按记录流方式读取的源，process_static_file_set()将其设置为有界批模式：只处理启动时已存在的文件，全部处理完后作业结束；monitor_continuously(interval)则相反，会持续监控新文件（参见 file_system.py）。WatermarkStrategy.for_monotonous_timestamps()为无时间戳的纯文本输入提供单调递增的水位线策略。批模式下该示例并没有真正使用事件时间，这里主要演示 API 的完整拼装方式。

方式二：from_collection读取内存数据。当未指定--input时，示例内置了莎士比亚《哈姆雷特》"To be, or not to be" 独白的 32 行文本作为默认数据（word_count.py），非常适合开箱即用地验证作业逻辑。

1.3 核心转换链：分词 → 映射 → 分组 → 归约

def split(line): yield from line.split() # compute word count ds = ds.flat_map(split) \ .map(lambda i: (i, 1), output_type=Types.TUPLE([Types.STRING(), Types.INT()])) \ .key_by(lambda i: i[0]) \ .reduce(lambda i, j: (i[0], i[1] + j[1]))

这一链式调用是 Word Count 的精髓，四个算子各司其职：

flat_map(split)：split是生成器函数，yield from line.split()将每一行按空白字符拆成单词并逐一发射，实现"一行 → 多词"的展平。
map(lambda i: (i, 1), output_type=...)：把每个单词映射为(单词, 1)二元组。output_type显式声明类型为Types.TUPLE([Types.STRING(), Types.INT()])，即字符串与整数的元组，这是 PyFlink 类型推断的重要环节——Lambda 表达式无法自动推断类型时，必须显式指定。
key_by(lambda i: i[0])：按单词本身分组，Flink 会依据 key 的哈希将相同单词路由到同一并行子任务，这是后续增量累加的前提。
reduce(lambda i, j: (i[0], i[1] + j[1]))：对同一 key 下的二元组两两归约，把计数累加。由于reduce是基于 key 的有状态算子，相同单词的计数会在其所在分区内持续累积，最终得到每个单词的总出现次数。

1.4 结果输出：FileSink 与 stdout 双通道

if output_path is not None: ds.sink_to( sink=FileSink.for_row_format( base_path=output_path, encoder=Encoder.simple_string_encoder()) .with_output_file_config( OutputFileConfig.builder() .with_part_prefix("prefix") .with_part_suffix(".ext") .build()) .with_rolling_policy(RollingPolicy.default_rolling_policy()) .build() ) else: print("Printing result to stdout. Use --output to specify output path.") ds.print()

FileSink.for_row_format(base_path, encoder)按行格式写出，Encoder.simple_string_encoder()将每条记录转为字符串后写入（参见 file_system.py）。
OutputFileConfig.builder().with_part_prefix("prefix").with_part_suffix(".ext")控制输出文件命名：分片文件会以prefix为前缀、以.ext为后缀。
RollingPolicy.default_rolling_policy()采用默认滚动策略，决定文件何时"滚动"成新文件（例如按文件大小或写入间隔，参见 file_system.py）。
若未指定--output，则调用ds.print()直接把结果打到标准输出，便于本地快速调试。

最后env.execute()提交作业执行。

1.5 命令行入口

if __name__ == '__main__': logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(message)s") parser = argparse.ArgumentParser() parser.add_argument('--input', dest='input', required=False, help='Input file to process.') parser.add_argument('--output', dest='output', required=False, help='Output file to write results to.') argv = sys.argv[1:] known_args, _ = parser.parse_known_args(argv) word_count(known_args.input, known_args.output)

使用argparse解析--input与--output两个可选参数，并用parse_known_args忽略 Flink 平台注入的其他参数。典型运行方式：

# 使用内置默认数据，结果打印到 stdout python pyflink/examples/datastream/word_count.py # 指定输入文件，结果写入输出目录 python pyflink/examples/datastream/word_count.py --input /path/to/input.txt --output /path/to/out

二、Streaming Word Count：无限数据流上的持续统计

流式示例 streaming_word_count.py 展示了另一个维度：数据源是一个持续产生数据的无限流，因此统计永远不会结束，这是与批处理版的本质区别。

2.1 基于 datagen 连接器构建无限数据源

words = ["flink", "window", "timer", "event_time", "processing_time", "state", "connector", "pyflink", "checkpoint", "watermark", "sideoutput", "sql", "datastream", "broadcast", "asyncio", "catalog", "batch", "streaming"] max_word_id = len(words) - 1

示例内置 18 个 Flink / PyFlink 领域词汇作为"字典"。数据源通过datagen 连接器构造：

env = StreamExecutionEnvironment.get_execution_environment() t_env = StreamTableEnvironment.create(stream_execution_environment=env) # define the source # randomly select 5 words per second from a predefined list t_env.create_temporary_table( 'source', TableDescriptor.for_connector('datagen') .schema(Schema.new_builder() .column('word_id', DataTypes.INT()) .build()) .option('fields.word_id.kind', 'random') .option('fields.word_id.min', '0') .option('fields.word_id.max', str(max_word_id)) .option('rows-per-second', '5') .build()) table = t_env.from_path('source') ds = t_env.to_data_stream(table)

这里出现了一个重要的跨 API 协作模式：

用StreamTableEnvironment.create(stream_execution_environment=env)在已有的 DataStream 执行环境之上创建 Table 环境。
通过TableDescriptor.for_connector('datagen')声明一个内置的 datagen 数据生成连接器表：字段word_id为INT类型，fields.word_id.kind=random表示随机取值，min=0、max=max_word_id限定取值范围，rows-per-second=5控制每秒生成 5 行数据，即每秒从词汇表中随机挑 5 个词。
最后t_env.to_data_stream(table)把 Table 转回 DataStream，完成从 Table API 到 DataStream API 的桥接，后续即可沿用 DataStream 的算子链做统计。

2.2 把 word_id 翻译成单词并统计

def id_to_word(r): # word_id is the first column of the input row return words[r[0]] # compute word count ds = ds.map(id_to_word) \ .map(lambda i: (i, 1), output_type=Types.TUPLE([Types.STRING(), Types.INT()])) \ .key_by(lambda i: i[0]) \ .reduce(lambda i, j: (i[0], i[1] + j[1]))

id_to_word取输入行的第一个字段（即word_id）作为下标，把数字翻译回词汇表中的单词；随后的map → key_by → reduce链与批处理版完全一致。由于数据源源源不断，reduce会持续维护每个单词的累计计数，任何时刻的状态都代表"到目前为止"的统计结果——这正是流式 Word Count 与批处理 Word Count 在语义上的关键差异。

2.3 输出与入口

if output_path is not None: ds.sink_to( sink=FileSink.for_row_format( base_path=output_path, encoder=Encoder.simple_string_encoder()) .with_output_file_config( OutputFileConfig.builder() .with_part_prefix("prefix") .with_part_suffix(".ext") .build()) .with_rolling_policy(RollingPolicy.default_rolling_policy()) .build() ) else: print("Printing result to stdout. Use --output to specify output path.") ds.print()

输出逻辑与批处理版共用同一套FileSink配置（行格式编码、prefix/.ext文件命名、默认滚动策略）。命令行仅提供--output一个可选参数：

# 默认打印到 stdout，观察持续刷新的计数结果 python pyflink/examples/datastream/streaming_word_count.py # 写入文件 python pyflink/examples/datastream/streaming_word_count.py --output /path/to/out

三、两个示例的对比与学习要点

维度	word_count.py（批处理）	streaming_word_count.py（流式）
运行模式	`RuntimeExecutionMode.BATCH`	默认流式语义
数据源	`FileSource`（文本文件）或`from_collection`（内存）	`datagen`连接器无限生成
数据是否有限	有限，处理完自动结束	无限，持续运行
统计语义	全量统计后输出一次	增量累计，状态持续演进
涉及额外 API	`WatermarkStrategy`、`StreamFormat`	`StreamTableEnvironment`、`TableDescriptor`、`to_data_stream`
输出方式	`FileSink`或`print()`	与批处理完全相同

两个示例共享的核心统计算子链（map → key_by → reduce）正是 DataStream 编程模型中最具代表性的模式：

key_by是状态与并行度的结合点：相同 key 的数据被路由到同一个并行实例，使reduce可以安全地在本地维护每个 key 的累加状态，无需全局协调。
显式output_type是 PyFlink 的实践要点：Lambda 表达式的返回类型无法被自动推断，必须用Types.TUPLE([Types.STRING(), Types.INT()])显式声明，否则作业提交阶段会因类型信息缺失而失败。

四、源码层面的纵深理解

4.1 FileSource 的批/流双模式

FileSourceBuilder（file_system.py）提供两个互斥的读取模式方法：

process_static_file_set()：有界模式，处理启动时路径下已存在的文件，全部完成后源即结束（示例批处理版使用）；
monitor_continuously(discovery_interval)：无界模式，按固定间隔扫描新文件并持续读取。

此外，StreamFormat.text_line_format()的源码注释揭示了两个底层事实（file_system.py）：

使用 Java 内置InputStreamReader按字符集解码，默认 UTF-8；
不支持 checkpoint 优化恢复：恢复时会重读并丢弃上次 checkpoint 之前已处理的行数，因为字符集解码器的内部缓冲状态无法精确定位行偏移。

而FileSource.for_record_stream_format还支持按文件扩展名自动解压.deflate、.xz、.bz2、.gz、.gzip等压缩格式（file_system.py），这让示例稍作改动即可直接处理压缩输入。

4.2 RuntimeExecutionMode 对行为的影响

execution_mode.py 对RuntimeExecutionMode的说明指出：运行模式不仅影响任务调度方式，还会影响网络 shuffle 行为、时间语义以及部分算子的记录发射行为。其中：

BATCH：任务先全部部署再执行，适合有界输入；
STREAMING：任务边部署边执行，开启 checkpoint，完整支持处理时间与事件时间，适合无界输入。

这也解释了为何批处理示例会主动set_parallelism(1)来保证输出单一文件——在批模式下并行度会显著影响分片文件的生成数量。

4.3 datagen 连接器的参数语义

流式示例通过TableDescriptor为 datagen 表配置了四个关键 option：

Option	示例值	含义
`fields.word_id.kind`	`random`	字段生成方式为随机值
`fields.word_id.min`	`0`	随机取值下限
`fields.word_id.max`	`17`（`len(words)-1`）	随机取值上限
`rows-per-second`	`5`	每秒生成的数据行数

由于是随机取值，每秒生成的 5 个词中必然存在重复，key_by + reduce的累计效果会随运行时间不断增长——这是演示流式增量统计最直观的方式。

五、运行前置条件与延伸阅读

运行这两个示例需要已安装 PyFlink 及其依赖（Py4J、CloudPickle、python-dateutil、Apache Beam 等，详见 flink-python/README.md）。本地开发可通过如下方式验证环境：

# 在仓库根目录下构建并安装 PyFlink 后即可运行示例 python flink-python/pyflink/examples/datastream/word_count.py

示例位于 flink-python/pyflink/examples/datastream 目录，同目录下还有 basic_operations.py（map/filter/key_by 基础操作）、process_json_data.py（JSON 处理）、state_access.py（状态访问）、event_time_timer.py（事件时间与定时器）、windowing（窗口）等进阶示例，对应文档索引见 DataStream 示例总览。这些示例与本文的 Word Count 共享同一套环境构建、算子链与 FileSink 输出模式，是继续深入 PyFlink DataStream API 的理想起点。

结语

从 word_count.rst 出发，本文完整还原了 PyFlink DataStream 的两个官方 Word Count 实现：批处理版展示了FileSource有界读取、内存集合输入与全量统计；流式版展示了 datagen 无限数据源、Table/DataStream 桥接与增量累计。两者共享的map → key_by → reduce算子链，是理解 Flink 状态化流处理的核心范式。掌握了这两个示例，你就拥有了构建更复杂 PyFlink 作业（窗口聚合、状态管理、多源连接）的坚实基础。