1. 从一个Word Count说起为什么它是大数据入门的必修课但凡接触过大数据处理框架的人几乎都绕不开Word Count这个案例。它就像编程语言里的Hello World看似简单实则涵盖了分布式计算最核心的几个概念数据读取、切分、转换、聚合、输出。你把这个案例吃透了后面学什么算子、什么窗口、什么状态管理心里都有底。我当初从传统数据处理转到大数据的头一个月就是被Flink的Word Count卡住的。倒不是逻辑有多复杂而是批处理和流处理两套API的写法差异、执行环境的区别、以及背后运行时的逻辑完全不同。网上很多教程要么只讲批不讲流要么代码跑不通要么版本对不上。所以这篇文章我打算把批处理和流处理两条路线都走一遍把踩过的坑和想明白的道理都写出来。这篇文章适合谁看如果你刚接触Flink或者之前用过其他计算框架想快速上手Flink再或者你已经在用Flink但一直没搞明白批流一体的真正含义那这篇内容应该能帮到你。我会从环境搭建开始到代码实现再到运行验证和常见问题排查一步步走完整个流程。代码基于Flink 1.17版本Java语言Maven构建这些都是目前企业里最主流的组合。核心关键词先摆出来Flink、批处理、流处理、Word Count、DataSet API、DataStream API、批流一体。这些词你会在全文反复看到每一个我都会掰开揉碎讲清楚。2. 动手之前环境准备与项目骨架搭建2.1 版本选型和依赖管理Flink的版本迭代很快不同版本之间API差异不小。我建议新手直接上Flink 1.17原因有几个首先这个版本对批流一体的支持已经非常成熟DataSet API虽然标记为遗留但仍然可用DataStream API在批模式下也能跑其次社区文档和生态适配都比较完善遇到问题容易找到答案。Maven依赖这块核心就两个包dependency groupIdorg.apache.flink/groupId artifactIdflink-java/artifactId version1.17.0/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version1.17.0/version /dependency如果你要在本地IDE里直接运行还需要加上flink-clients依赖。很多教程漏掉这个导致代码编译通过但一运行就报No ExecutorFactory found的错误。这个坑我踩过排查了半天才发现是缺包。dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version1.17.0/version /dependency注意Flink 1.17中flink-java这个包主要提供DataSet API相关的类而flink-streaming-java提供DataStream API。两个都要加因为我们要同时演示批处理和流处理两种实现。2.2 项目目录结构规划我习惯把批处理和流处理的代码分开放这样结构清晰也方便对比flink-wordcount-demo/ ├── pom.xml ├── src/ │ └── main/ │ ├── java/ │ │ └── com/ │ │ └── demo/ │ │ ├── batch/ │ │ │ └── BatchWordCount.java │ │ └── stream/ │ │ ├── StreamWordCount.java │ │ └── StreamWordCountWithLambda.java │ └── resources/ │ └── input/ │ └── words.txtwords.txt就是我们的测试数据源内容随便写几行英文文本就行比如hello flink hello world flink batch processing stream processing hello flink word count demo这个文件后面批处理和流处理都会用到批处理直接读文件流处理可以用nc命令模拟实时数据也可以读文件模拟流。2.3 IDE和运行环境的小细节我用的是IntelliJ IDEA这里有个小细节要注意Flink程序在IDE里运行时默认的并行度是CPU核数。如果你的机器核数比较多本地跑Word Count会看到输出文件被分成多个part文件每个文件里是部分结果。这不是bug是并行执行的正常表现。如果你想看到单一输出文件可以在代码里设置并行度为1。另外Java版本建议用Java 8或Java 11Flink 1.17对这两个版本支持最好。Java 17虽然也能跑但偶尔会有模块化相关的警告新手容易慌。3. 批处理实现DataSet API的经典写法3.1 批处理的核心思路拆解批处理的逻辑很直观一次性把数据全部读进来然后做切分、映射、分组、聚合最后输出结果。你可以把它想象成用Excel处理一个几万行的表格——数据是静态的、有界的处理完就结束了。Flink的DataSet API就是为这种场景设计的。核心步骤是获取ExecutionEnvironment批处理执行环境用readTextFile读取数据源用flatMap做切分和转换把每行文本拆成一个个单词每个单词记为1用groupBy按单词分组用sum对每组求和用writeAsText输出结果调用execute触发执行这个流程和MapReduce的思路几乎一样但写起来简洁太多了。MapReduce你要写Mapper、Reducer、Driver三个类Flink一个类就搞定。3.2 完整代码实现与逐行注释package com.demo.batch; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.java.DataSet; import org.apache.flink.api.java.ExecutionEnvironment; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.util.Collector; public class BatchWordCount { public static void main(String[] args) throws Exception { // 1. 创建批处理执行环境 ExecutionEnvironment env ExecutionEnvironment.getExecutionEnvironment(); // 2. 设置并行度为1方便本地查看完整结果 env.setParallelism(1); // 3. 读取本地文件作为数据源 String inputPath src/main/resources/input/words.txt; DataSetString textLines env.readTextFile(inputPath); // 4. 核心转换逻辑切分、映射、分组、聚合 DataSetTuple2String, Integer wordCounts textLines .flatMap(new FlatMapFunctionString, Tuple2String, Integer() { Override public void flatMap(String line, CollectorTuple2String, Integer out) { // 按空格切分每行文本 String[] words line.toLowerCase().split(\\s); for (String word : words) { if (word.length() 0) { // 每个单词输出一个(单词, 1)的二元组 out.collect(new Tuple2(word, 1)); } } } }) // 按单词字段f0分组 .groupBy(0) // 对计数字段f1求和 .sum(1); // 5. 输出结果到文件 wordCounts.writeAsText(src/main/resources/output/batch_result); // 6. 触发执行 env.execute(Batch Word Count Demo); } }这段代码有几个地方值得展开说。flatMap和map的区别map是一进一出flatMap是一进多出。我们读进来的一行文本包含多个单词需要拆成多个输出所以用flatMap。这个算子是Flink里最常用的转换之一务必理解清楚。Tuple2是Flink内置的二元组类型f0是第一个字段单词f1是第二个字段计数。groupBy(0)表示按第一个字段分组sum(1)表示对第二个字段求和。这种基于位置的字段引用是DataSet API的典型写法写起来简洁但可读性一般字段多了容易搞混。env.execute()是必须调用的不调用的话整个流程不会执行。这一点和Spark的惰性求值类似转换算子只是构建了执行计划真正触发计算的是execute。3.3 运行结果验证与输出解读跑完上面的代码你会在output/batch_result目录下看到一个文件内容大概是(flink,3) (hello,3) (world,1) (batch,1) (processing,2) (stream,1) (word,1) (count,1) (demo,1)这里有个细节输出结果默认是按单词分组的但组与组之间的顺序不保证。因为Flink内部是并行处理的不同分区的结果合并时顺序可能变化。如果你需要有序输出可以在后面加一个sortPartition或者收集到本地再排序。还有一个容易困惑的点writeAsText输出的文件可能不止一个。如果你没设置并行度为1每个并行子任务会输出一个part文件。这是分布式计算的正常行为不是错误。实操心得本地调试时把并行度设为1输出结果一目了然。但到了生产环境并行度要根据数据量和集群资源来定一般设置为CPU核数的2到4倍比较合适。4. 流处理实现DataStream API的实时玩法4.1 流处理与批处理的本质差异批处理处理的是有界数据流处理处理的是无界数据。这个有界和无界的区别决定了两者在API设计、执行模式、结果输出上的根本不同。批处理知道数据什么时候结束所以可以做全局排序、全局聚合。流处理不知道数据什么时候结束所以只能用窗口来切分计算范围或者用状态来累积结果。Word Count这个案例里流处理每来一条数据就更新一次计数结果是一个持续变化的流。Flink的DataStream API就是为流处理设计的。和DataSet API相比最大的区别是DataStream API在批模式下也能运行这就是批流一体的体现。同一套代码换个执行模式就能处理有界数据。4.2 基于DataStream API的完整实现package com.demo.stream; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.util.Collector; public class StreamWordCount { public static void main(String[] args) throws Exception { // 1. 创建流处理执行环境 StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); // 2. 设置并行度 env.setParallelism(1); // 3. 读取数据源这里用socket模拟实时数据流 DataStreamString textStream env.socketTextStream(localhost, 9999); // 4. 核心转换逻辑 DataStreamTuple2String, Integer wordCounts textStream .flatMap(new FlatMapFunctionString, Tuple2String, Integer() { Override public void flatMap(String line, CollectorTuple2String, Integer out) { String[] words line.toLowerCase().split(\\s); for (String word : words) { if (word.length() 0) { out.collect(new Tuple2(word, 1)); } } } }) // 按单词分组 .keyBy(value - value.f0) // 对计数求和 .sum(1); // 5. 输出到控制台 wordCounts.print(); // 6. 触发执行 env.execute(Stream Word Count Demo); } }和批处理版本对比有几个关键差异ExecutionEnvironment换成了StreamExecutionEnvironment这是两个不同的执行环境分别对应批和流。readTextFile换成了socketTextStream数据源从静态文件变成了实时socket流。当然你也可以用readTextFile在流环境里读文件Flink会把它当作有界流处理。groupBy换成了keyBy。groupBy是DataSet API的分组算子keyBy是DataStream API的分组算子。功能类似但keyBy返回的是KeyedStream后续可以接窗口操作这是流处理特有的。writeAsText换成了print。流处理的结果是持续输出的写到文件里不太直观直接打印到控制台更方便观察。4.3 用Lambda表达式简化代码Java 8的Lambda表达式可以让代码更简洁DataStreamTuple2String, Integer wordCounts textStream .flatMap((String line, CollectorTuple2String, Integer out) - { for (String word : line.toLowerCase().split(\\s)) { if (word.length() 0) { out.collect(new Tuple2(word, 1)); } } }) .returns(Types.TUPLE(Types.STRING, Types.INT)) .keyBy(value - value.f0) .sum(1);这里有个坑Lambda表达式在Flink里会有类型擦除的问题所以必须加.returns(Types.TUPLE(Types.STRING, Types.INT))显式声明返回类型否则运行时会报类型推断失败的异常。这个错误信息不太友好新手很容易卡在这里。注意Lambda表达式虽然简洁但调试时堆栈信息不如匿名内部类清晰。生产环境我建议还是用匿名内部类或者单独定义Function类方便排查问题。4.4 本地测试流处理的完整流程流处理需要数据源持续产生数据本地测试最方便的方式是用nc命令模拟socket流。Linux和Mac自带ncWindows可以用ncat或者PowerShell的Test-NetConnection替代。启动nc监听9999端口nc -lk 9999然后运行StreamWordCount程序在nc终端里逐行输入文本hello flink hello world flink stream processing每输入一行程序的控制台就会输出更新后的计数结果(hello,1) (flink,1) (hello,2) (world,1) (flink,2) (stream,1) (processing,1)这就是流处理的魅力结果随着数据到来实时更新而不是等所有数据到齐再计算。5. 批流一体的真正含义同一套代码两种模式5.1 Flink的批流一体设计哲学Flink最引以为傲的特性之一就是批流一体。但很多人对这个概念的理解停留在批处理是流处理的特例这个层面其实远不止如此。在Flink 1.12之前批处理和流处理是两套完全独立的APIDataSet API和DataStream API。两套API的算子不通用代码不能互相迁移。从1.12开始Flink提出了批流一体的战略核心思路是用DataStream API统一批和流通过设置执行模式来切换。具体来说StreamExecutionEnvironment可以设置三种执行模式RuntimeExecutionMode.STREAMING纯流模式默认RuntimeExecutionMode.BATCH批模式用于有界数据RuntimeExecutionMode.AUTOMATIC自动根据数据源是否有界来选择这意味着你可以用同一套DataStream API代码处理有界数据时自动切换到批模式享受批处理的优化处理无界数据时用流模式保证实时性。5.2 用DataStream API跑批处理把上面的流处理代码稍微改一下就能跑批处理StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setRuntimeMode(RuntimeExecutionMode.BATCH); env.setParallelism(1); DataStreamString textStream env.readTextFile(src/main/resources/input/words.txt); DataStreamTuple2String, Integer wordCounts textStream .flatMap(new FlatMapFunctionString, Tuple2String, Integer() { Override public void flatMap(String line, CollectorTuple2String, Integer out) { for (String word : line.toLowerCase().split(\\s)) { if (word.length() 0) { out.collect(new Tuple2(word, 1)); } } } }) .keyBy(value - value.f0) .sum(1); wordCounts.print(); env.execute(Unified Word Count Demo);注意这里用的是readTextFile而不是socketTextStream因为批模式要求数据源是有界的。如果数据源是无界的批模式会报错。5.3 两种模式的性能差异与适用场景批模式和流模式在底层执行上有本质区别。批模式会对执行计划做更多优化比如算子融合、数据分区优化、排序合并等吞吐量更高。流模式则更注重低延迟数据一来就处理不做额外的优化等待。我用同样的数据量测试过批模式的执行时间大约是流模式的60%到70%。但流模式的优势在于实时性数据延迟可以控制在毫秒级。选择哪种模式取决于你的业务需求对比维度批处理模式流处理模式数据特征有界数据无界数据延迟要求分钟级到小时级毫秒级到秒级吞吐量高中到高结果输出一次性输出持续更新典型场景离线报表、数据清洗实时监控、告警实操心得如果你的业务对延迟不敏感但数据量大、计算复杂优先用批模式。如果业务要求实时响应哪怕数据量不大也要用流模式。不要为了追求技术先进而强行用流处理做离线任务那是给自己找麻烦。6. 常见问题与排查技巧实录6.1 编译和运行时的典型报错问题一No ExecutorFactory found to execute the application这个错误几乎每个Flink新手都会遇到。原因是缺少flink-clients依赖。在IDE里运行时Flink需要这个包来创建本地执行器。加上依赖重新编译即可。问题二The generic type parameters of Collector are missing这是Lambda表达式的类型擦除问题。解决方法是在flatMap后面加.returns(Types.TUPLE(Types.STRING, Types.INT))显式声明返回类型。问题三java.net.ConnectException: Connection refused流处理连接socket失败。检查nc命令是否已经启动端口是否被占用。可以先用telnet localhost 9999测试端口连通性。问题四输出目录已存在导致作业失败writeAsText指定的输出目录如果已经存在Flink会报错。每次运行前手动删除输出目录或者在代码里加上删除逻辑。6.2 性能调优的实用技巧并行度设置是影响性能最直接的因素。本地测试设1就行生产环境一般设为CPU核数的2到4倍。但也不是越大越好并行度太高会导致任务调度开销增加反而降低吞吐量。数据倾斜是另一个常见问题。如果某个单词出现频率特别高处理这个单词的子任务就会成为瓶颈。解决办法是在keyBy之前加一个随机前缀做两阶段聚合。这个技巧在真实业务里非常实用比如统计热门商品点击量时热门商品的key就会倾斜。检查点配置对容错很重要。流处理模式下开启检查点可以保证故障恢复时不丢数据。但检查点会增加一定的性能开销需要根据业务对一致性的要求来权衡。6.3 常见问题速查表问题现象可能原因解决方法编译报错找不到类依赖缺失或版本不匹配检查pom.xml确保flink-java、flink-streaming-java、flink-clients版本一致运行时类型推断失败Lambda表达式类型擦除加.returns()显式声明类型连接socket失败nc未启动或端口占用检查nc进程换端口重试输出目录已存在上次运行结果未清理删除输出目录或代码中加清理逻辑结果顺序不一致并行处理导致设置并行度为1或加排序算子作业提交后无输出未调用execute检查是否调用了env.execute()内存溢出数据量过大或状态未清理增加内存配置设置状态TTL7. 从Word Count延伸出去还能学什么Word Count虽然简单但它是理解Flink核心概念的入口。把这个案例吃透之后你可以沿着几条线继续深入。第一条线是窗口计算。流处理里数据是无限的要统计最近5分钟的单词出现次数就需要用到窗口。Flink提供了滚动窗口、滑动窗口、会话窗口等多种窗口类型每种窗口的适用场景不同。第二条线是状态管理。Word Count里的计数其实就是一种状态。Flink提供了丰富的状态类型和状态后端可以管理大规模的状态数据。理解状态是理解Flink容错机制的基础。第三条线是时间语义。Flink支持事件时间、处理时间、摄入时间三种时间语义。事件时间处理乱序数据的能力是Flink的杀手锏也是流处理里最复杂的部分。第四条线是连接器生态。Word Count用的是文件和socket真实业务里数据来自Kafka、数据库、消息队列等。Flink提供了丰富的连接器可以和各种数据源和存储系统对接。我个人在实际操作中的体会是Word Count这个案例值得反复写、反复改。第一遍照着教程写第二遍不看教程自己写第三遍尝试用不同的API写第四遍加上窗口和状态。每写一遍对Flink的理解都会深一层。踩过几次坑之后你会发现那些曾经让你头疼的报错其实都是在帮你理解框架的设计逻辑。