Flink批流双模式Word Count实战：从入门到避坑

1. 从一个Word Count说起为什么它是Flink的Hello World但凡接触过大数据处理的人对Word Count都不会陌生。它就像编程语言里的Hello World看似简单却涵盖了数据处理框架最核心的几个概念数据读取、转换、分组、聚合、输出。我刚开始接触Flink的时候也是从一个Word Count案例入手的当时觉得这玩意儿能有多难结果真正跑起来才发现批处理和流处理两种模式下代码写法、执行逻辑、结果输出都有不少门道。这篇文章要聊的就是如何用Flink分别以批处理和流处理两种方式实现Word Count。你可能会问一个Word Count有什么好讲的网上不是一搜一大把吗说实话确实很多但大部分要么只给代码不讲原理要么版本太老跑不起来要么批和流混在一起讲得云里雾里。我踩过的坑包括但不限于依赖版本冲突导致NoSuchMethodError、流处理模式下文件读取完程序不退出、批处理模式下ExecutionEnvironment和StreamExecutionEnvironment搞混导致编译报错等等。所以这篇内容适合以下人群刚接触Flink想跑通第一个案例的新手、从Spark Streaming或其他框架转过来想快速上手Flink的开发者、以及想搞清楚批流两种模式底层差异的进阶学习者。我会从环境搭建开始把两种模式的代码逐行拆解补充参数配置、执行原理、常见报错和排查思路让你看完就能自己跑起来而且知道每一步为什么这么写。提示本文基于Flink 1.17.x版本编写不同版本API可能有差异尤其是1.12之前和之后的版本在DataStream API上有较大变化建议尽量使用1.14以上的稳定版本。2. 环境准备与项目骨架搭建2.1 开发环境选型和依赖管理工欲善其事必先利其器。Flink开发的环境准备其实不复杂但有几个关键点容易翻车。首先是JDK版本。Flink 1.17要求JDK 8或11我实测下来JDK 11最稳JDK 8也能跑但某些新API会有警告。JDK 17目前兼容性还不够好不建议新手折腾。你可以用java -version确认一下当前版本。其次是构建工具。Maven和Gradle都行但社区里Maven的示例更多遇到问题好搜解决方案所以我这里用Maven演示。创建一个标准的Maven项目pom.xml里核心依赖就两个flink-streaming-java和flink-clients。注意如果你要跑批处理不需要额外引入批处理的依赖因为Flink 1.12之后批处理统一用DataStream API的ExecutionEnvironment来做了DataSet API已经进入维护状态官方建议新项目直接用DataStream API的批模式。properties flink.version1.17.1/flink.version java.version11/java.version /properties dependencies dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink.version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients/artifactId version${flink.version}/version /dependency /dependencies这里有个坑要提醒很多人会习惯性地加上flink-java依赖觉得批处理肯定需要它。实际上从1.12开始flink-java里的DataSet API已经不推荐使用了加了反而可能引起依赖冲突。我自己就遇到过同时引入flink-java和flink-streaming-java后ExecutionEnvironment类加载到了错误的版本编译能过但运行时报ClassNotFoundException。2.2 输入数据准备与项目目录结构Word Count的输入数据很简单一个文本文件就行。我在项目根目录下建了个input文件夹放了一个words.txt内容随便写几行英文单词比如hello flink hello world flink batch and stream word count example hello hello flink项目结构大概是这样flink-wordcount/ ├── pom.xml ├── input/ │ └── words.txt └── src/ └── main/ └── java/ └── com/ └── example/ ├── BatchWordCount.java └── StreamWordCount.java注意输入文件路径在代码里写相对路径时要搞清楚是相对于项目根目录还是相对于运行时的classpath。IDEA里直接Run main方法工作目录默认是项目根目录所以input/words.txt能直接找到。但如果你打成jar包提交到集群路径就得改成绝对路径或者HDFS路径这个后面会细说。3. 批处理模式实现Word Count3.1 批处理代码逐行拆解先上完整代码然后我逐段解释为什么这么写。package com.example; import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; public class BatchWordCount { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setRuntimeMode(org.apache.flink.api.common.RuntimeExecutionMode.BATCH); DataStreamString text env.readTextFile(input/words.txt); DataStreamTuple2String, Integer counts text .flatMap((String line, org.apache.flink.util.CollectorTuple2String, Integer out) - { for (String word : line.split( )) { out.collect(Tuple2.of(word, 1)); } }) .returns(Types.TUPLE(Types.STRING, Types.INT)) .keyBy(value - value.f0) .sum(1); counts.print(); env.execute(Batch WordCount); } }第一行StreamExecutionEnvironment.getExecutionEnvironment()获取执行环境。这里你可能注意到批处理用的也是StreamExecutionEnvironment而不是ExecutionEnvironment。这就是Flink 1.12之后的变化——批流统一用DataStream API通过setRuntimeMode来指定是BATCH还是STREAMING模式。如果你在网上看到用ExecutionEnvironment和DataSet的代码那是老版本的写法虽然还能跑但已经不推荐了。env.setRuntimeMode(RuntimeExecutionMode.BATCH)这行是关键。它告诉Flink以批处理模式执行底层会用更高效的批处理优化策略比如对排序、分组操作做特殊优化而不是像流处理那样逐条处理。env.readTextFile(input/words.txt)读取文本文件返回一个DataStreamString每一行是一个元素。接下来是核心的转换逻辑。flatMap把每一行拆成单词并输出(word, 1)的二元组。这里有个细节Java的Lambda表达式在Flink里会有类型擦除问题所以必须加.returns(Types.TUPLE(Types.STRING, Types.INT))显式声明返回类型否则运行时会报InvalidTypesException。这个坑我踩过不止一次尤其是从Scala转Java的人特别容易忘。keyBy(value - value.f0)按单词分组sum(1)对第二个字段求和。注意keyBy返回的是KeyedStreamsum之后又变回DataStream。最后counts.print()把结果打印到控制台env.execute()触发实际执行。Flink是懒执行的前面所有的转换操作只是构建了执行图只有调用execute()才会真正提交任务。3.2 批处理执行原理与输出解读跑起来之后控制台输出大概是这样(flink,3) (hello,4) (world,1) (batch,1) (and,1) (stream,1) (word,1) (count,1) (example,1)你会发现输出顺序和输入顺序不一致。这是因为批处理模式下Flink会对数据进行分组和排序优化相同key的数据会被分配到同一个分区处理最终输出的顺序取决于分区和合并的策略。这不是bug是批处理的正常行为。批处理模式的核心特点是有界数据、全量计算、结果一次性输出。Flink在批模式下会尽可能优化执行计划比如对keyBy后的数据做排序而不是哈希减少内存压力。如果你处理的是历史日志分析、离线报表这类场景批模式是首选。实操心得批处理模式下print()的输出会带上线程ID前缀如1这是并行执行的标识。如果你想让结果按单词字母序输出可以在sum之后加一个.sortPartition(0, Order.ASCENDING)但注意这会产生额外的排序开销数据量大时慎用。4. 流处理模式实现Word Count4.1 流处理代码与批处理的差异对比流处理的代码和批处理长得很像但有几个关键区别。package com.example; import org.apache.flink.api.common.typeinfo.Types; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; public class StreamWordCount { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); DataStreamString text env.readTextFile(input/words.txt); DataStreamTuple2String, Integer counts text .flatMap((String line, org.apache.flink.util.CollectorTuple2String, Integer out) - { for (String word : line.split( )) { out.collect(Tuple2.of(word, 1)); } }) .returns(Types.TUPLE(Types.STRING, Types.INT)) .keyBy(value - value.f0) .sum(1); counts.print(); env.execute(Stream WordCount); } }乍一看几乎一样区别在于没有setRuntimeMode(BATCH)默认就是STREAMING模式加了env.setParallelism(1)把并行度设为1方便观察输出顺序。流处理模式下readTextFile读取的是有界数据但Flink会把它当作流来处理——逐行读取、逐条处理、每来一条数据就更新结果并输出。所以你会看到输出是这样的(hello,1) (flink,1) (hello,2) (world,1) (hello,3) (flink,2) ...每处理一个单词就输出一次当前累计结果这就是流处理的增量计算特性。和批处理的全量计算不同流处理是来一条算一条结果实时更新。4.2 流处理模式的核心机制解析流处理模式有几个批处理没有的概念理解它们对后续深入学习很重要。并行度与任务槽。setParallelism(1)意味着所有操作符都在一个线程里串行执行。如果不设置默认并行度是CPU核数输出会交错在一起看起来比较乱。实际生产环境中并行度根据数据量和集群资源来定一般设置为Kafka分区数的整数倍比较合理。KeyBy的哈希分区。流处理模式下keyBy会对key做哈希然后对并行度取模决定数据发往哪个分区。相同key的数据一定在同一个分区这样才能正确累加。批处理模式下虽然逻辑一样但底层实现可能用排序代替哈希这是Flink的内部优化。事件时间与水位线。这个Word Count案例用的是处理时间Processing Time也就是数据被处理的系统时间。实际流处理场景中数据可能乱序到达就需要用事件时间Event Time加水位线Watermark来处理。不过这是进阶内容入门阶段先不用管。注意流处理模式下如果读取的是有界文件程序处理完所有数据后会自动退出。但如果读取的是Kafka这类无界数据源程序会一直运行需要用CtrlC或者提交取消命令来停止。我见过有同学跑流处理Word Count读的是socket数据源结果程序不退出以为代码写错了其实是正常行为。5. 批流两种模式的深度对比与选型建议5.1 执行模式、性能与适用场景对比很多人学完批和流两个版本后最大的困惑是到底什么时候用哪个我把关键差异整理成了一张表方便你对照。对比维度批处理模式流处理模式数据特征有界数据文件、历史表无界数据消息队列、实时日志计算方式全量计算一次性输出增量计算持续输出执行模式RuntimeExecutionMode.BATCHRuntimeExecutionMode.STREAMING延迟分钟级到小时级毫秒级到秒级结果输出最终结果每次更新的中间结果资源占用相对较高需要排序、缓存相对较低逐条处理典型场景离线报表、历史数据分析实时监控、实时推荐、告警从性能角度看批处理模式在处理有界数据时通常更快因为Flink会做执行计划优化比如算子融合、排序合并等。流处理模式胜在低延迟但吞吐量可能不如批处理因为每条数据都要走完整的处理链路。选型的核心判断标准就一条你的数据有没有边界业务对延迟的要求是什么。如果数据是有限的、可以等几分钟甚至几小时出结果用批处理。如果数据源源不断、需要秒级甚至毫秒级响应用流处理。5.2 代码复用与批流一体的实践思路既然两种模式代码这么像能不能写一份代码同时支持批和流答案是肯定的这也是Flink批流一体理念的体现。核心思路是把setRuntimeMode做成可配置的通过命令行参数或者配置文件来决定用哪种模式。数据源也可以抽象批处理用readTextFile流处理用addSource比如Kafka Source但后续的转换逻辑完全复用。// 伪代码示意 StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); if (isBatch) { env.setRuntimeMode(RuntimeExecutionMode.BATCH); text env.readTextFile(path); } else { env.setRuntimeMode(RuntimeExecutionMode.STREAMING); text env.addSource(kafkaSource); } // 后续转换逻辑完全一样这样做的好处是维护一套代码降低开发和测试成本。不过要注意批和流的语义差异还是存在的比如流处理中的状态管理、水位线机制在批处理中不存在复杂场景下还是需要分别处理。实操心得我在实际项目中会先把批处理版本跑通确认逻辑正确再改成流处理版本。因为批处理调试更方便结果确定容易验证。流处理涉及时间语义和状态调试成本高很多。这个顺序反过来会浪费很多时间。6. 常见报错与排查技巧实录6.1 依赖冲突与类型擦除问题问题一NoSuchMethodError或ClassNotFoundException这是最典型的依赖冲突。表现是编译能过运行时报找不到类或方法。原因通常是Flink版本不一致或者引入了多个版本的Flink依赖。排查方法用mvn dependency:tree查看依赖树确认所有Flink相关依赖版本一致。如果引入了flink-java和flink-streaming-java检查版本号是否相同。另外flink-clients在1.17里是必须的不加会报找不到StreamExecutionEnvironment的实现类。问题二InvalidTypesException: The generic type parameters of Collector are missing这是Lambda类型擦除导致的。Flink无法从Lambda表达式推断出返回类型必须用.returns()显式声明。解决方法在flatMap后面加.returns(Types.TUPLE(Types.STRING, Types.INT))。如果是其他类型用Types.POJO(YourClass.class)或TypeInformation.of(YourClass.class)。问题三keyBy之后sum结果不对检查keyBy的字段索引是否正确。Tuple2的f0是第一个字段f1是第二个字段。sum(1)表示对第二个字段求和。如果写成了sum(0)会对字符串求和直接报错。6.2 运行环境与并行度相关故障问题四输出结果顺序混乱这是并行度导致的。默认并行度是CPU核数多个线程同时输出顺序自然乱。解决方法env.setParallelism(1)或者接受乱序因为分布式计算本来就不保证全局有序。问题五流处理程序不退出如果数据源是有界文件处理完会自动退出。如果是Kafka或Socket等无界源程序会一直运行。这是正常行为不是bug。需要手动停止或通过Flink的取消机制停止。问题六文件路径找不到readTextFile的路径在本地运行时相对于项目根目录提交到集群时相对于TaskManager的工作目录。建议用绝对路径或HDFS路径。如果路径是file:///...开头表示本地文件系统如果是hdfs://...表示HDFS。报错信息可能原因解决方法NoSuchMethodError依赖版本冲突统一Flink版本检查依赖树InvalidTypesExceptionLambda类型擦除加.returns()显式声明类型ClassNotFoundException缺少flink-clients依赖添加flink-clients依赖结果顺序混乱并行度大于1setParallelism(1)或接受乱序程序不退出无界数据源正常行为手动停止文件找不到路径问题用绝对路径或HDFS路径避坑技巧我习惯在main方法第一行加env.setParallelism(1)调试阶段先保证结果可预测等逻辑验证无误后再调大并行度做性能测试。这个习惯帮我省了很多排查时间。7. 从Word Count延伸下一步可以学什么Word Count跑通之后其实只是打开了Flink的大门。基于这个案例你可以往几个方向继续深入。第一个方向是换数据源。把readTextFile换成Kafka Source体验真正的流处理。你会接触到KafkaSource的配置、偏移量管理、水位线生成等概念。这个过程中Word Count的逻辑不变但数据接入方式变了能帮你理解Flink的数据源抽象。第二个方向是加时间语义。给数据加上时间戳用事件时间代替处理时间配合水位线处理乱序数据。这是流处理的核心难点也是面试高频考点。你可以把输入数据改成带时间戳的格式然后观察窗口计算的结果变化。第三个方向是状态管理。Word Count里的sum其实是有状态的Flink帮你管理了状态。你可以进一步学习KeyedState、OperatorState了解状态后端Memory、Fs、RocksDB的区别和选型。数据量大时状态后端的选择直接影响性能和稳定性。第四个方向是容错机制。了解Checkpoint和Savepoint的原理知道Flink怎么保证Exactly-Once语义。这个在Word Count案例里体现不出来但生产环境必不可少。我自己是从Word Count开始然后逐步加了Kafka源、事件时间、窗口、状态最后搭了一个完整的实时统计Demo。每一步都会遇到新问题但每一步也都能加深对Flink的理解。建议你不要急着上复杂场景先把Word Count的批和流两个版本吃透把依赖、类型、并行度这些基础问题搞清楚后面会顺很多。最后分享一个小技巧Flink的Web UI是个好东西。本地运行时会启动一个Web界面默认8081端口可以看到执行计划、算子链、数据流量、背压情况。跑Word Count的时候打开看看能直观理解Flink是怎么把代码变成执行图的。这个习惯我从入门保持到现在排查问题时特别有用。