Flink本地开发环境搭建与WordCount实战：从零跑通流处理

1. 为什么要在本地IDE里搭一套Flink开发环境

刚接触流式计算那会儿，我最常干的一件事就是抱着官方文档翻来覆去地看，看完觉得“哦，原来是这样”，然后一合上文档，连个WordCount都跑不起来。后来带过几个新人，发现大家卡住的地方几乎一模一样：不是算子不会写，而是环境没搭对，依赖冲突、Scala版本对不上、日志刷不出来、任务提交了没结果。所以这个系列的第一篇，我不打算聊什么高深的状态管理或者Exactly-Once，就老老实实把本地开发环境从零搭起来，再跑通一个WordCount，把这条链路彻底走顺。

Flink这套东西，本质上是一个分布式的流批一体计算引擎。你可以把它想象成一个“数据流水线工厂”：数据从一头进来，经过一道道工序（算子）加工，从另一头出去。而我们要做的，就是在自己的电脑上先建一个“迷你工厂”，用IDE当控制台，用本地JVM当集群，把代码写完直接点运行就能看到结果。这套环境搭好之后，后面学窗口、学状态、学Checkpoint，全都能在这个基础上快速验证，不用每次都去折腾服务器。

这篇文章适合谁看？如果你是会写Java、用过Maven、但对Flink还比较陌生，想在自己电脑上把第一个流处理程序跑起来，那这篇就是给你准备的。如果你已经用过Flink但一直用公司现成的环境，想搞清楚本地开发到底该怎么配，也能从里面找到一些踩坑经验。我会把版本选择、依赖配置、代码结构、运行方式、常见报错都讲清楚，尽量让你照着做就能复现。

2. 环境搭建前的整体思路与版本选型

2.1 先想清楚：本地开发环境和生产集群的区别

很多人一上来就问“Flink装哪个版本”，其实这个问题得先分清楚你要干什么。本地开发环境的核心目标是快速验证逻辑，不是模拟生产压力。所以本地环境有几个特点：用本地JVM进程模拟JobManager和TaskManager，用IDE直接运行main方法，依赖通过Maven管理，日志打到控制台。而生产集群要考虑高可用、资源调度、状态后端、网络传输，那是另一回事。

我见过有新人非要在本地搭一个Standalone集群，然后从IDE提交任务过去，结果光调网络和端口就耗掉一整天。不是说不行，而是没必要。对于学习和日常开发，IDE直接运行是最省事的路径。等你把算子逻辑都摸熟了，再去折腾集群提交也不迟。

所以这篇文章的路线是：Maven项目 + IDE直接运行 + 本地Flink依赖。这条路线的好处是启动快、调试方便、断点能打、日志直接看。缺点是它不涉及分布式细节，但这恰恰是入门阶段应该先放一放的东西。

2.2 版本怎么选：Flink、Scala、Java三者的关系

Flink的版本选择，核心是看三件事：Flink版本、Scala版本、Java版本。这三者之间有兼容性要求，选错了就是各种NoSuchMethodError和ClassNotFoundException。

先说Flink版本。截至我写这篇内容的时候，Flink 1.17和1.18是比较稳的长期支持方向，1.19、1.20也有不少人在用。对于新手，我建议选1.17.x或1.18.x，原因是这两个版本的文档全、社区案例多、和各类连接器的兼容性经过充分验证。太新的版本有时候连接器还没跟上，太老的版本又缺少一些新特性。

再说Scala版本。Flink的Java API和Scala API是分开的，如果你用Java写，其实Scala版本只影响Flink内部依赖的Scala库版本。Flink 1.15之后，官方主推的Scala版本是2.12，部分版本也支持2.11。我的建议是统一用Scala 2.12，因为大部分连接器和示例都是基于2.12的。如果你用Scala写业务代码，那更要注意版本一致，否则会出现序列化相关的诡异问题。

Java版本这块，Flink 1.17及以前对Java 8支持很好，Java 11也OK。Flink 1.18开始对Java 17的支持更完善。我个人的习惯是用Java 8或Java 11，因为这两个版本在各类依赖里兼容性最好，不容易碰到模块化带来的反射问题。如果你非要用Java 17，记得加一些JVM参数开放模块访问，否则Flink的反射机制会报错。

下面这张表是我总结的版本搭配建议，可以直接参考：

提示：不要同时在一个项目里混用多个Scala版本，Maven依赖树里出现两个scala-library版本时，运行期报错会非常难查。

2.3 构建工具选Maven还是Gradle

这个问题其实没有标准答案，但我观察下来，Flink社区里Maven的示例和文档占绝大多数，官方QuickStart也是Maven。对于新手，用Maven能少走很多弯路，因为你能直接抄官方pom里的依赖配置，遇到问题搜出来的答案也大多是Maven的。

Gradle当然也能用，而且构建速度更快、脚本更灵活，但如果你对Gradle不熟，光是配Flink的依赖范围（provided、compile、runtime）就够折腾一阵。所以这篇文章以Maven为主线，Gradle用户可以把依赖配置对应翻译过去，逻辑是一样的。

3. 手把手搭建Maven项目与依赖配置

3.1 创建项目骨架：从archetype到目录结构

最省事的创建方式是用Flink官方提供的Maven archetype，一条命令就能生成标准目录结构。命令大概长这样：

mvn archetype:generate \ -DarchetypeGroupId=org.apache.flink \ -DarchetypeArtifactId=flink-quickstart-java \ -DarchetypeVersion=1.17.2 \ -DgroupId=com.example.flink \ -DartifactId=flink-wordcount-demo \ -Dversion=1.0-SNAPSHOT \ -Dpackage=com.example.flink

执行完之后，你会得到一个标准Maven项目，目录结构大致是：

flink-wordcount-demo ├── pom.xml └── src └── main ├── java │ └── com/example/flink │ ├── BatchJob.java │ └── StreamingJob.java └── resources └── log4j2.properties

这个骨架里已经包含了Flink的核心依赖和一个流处理、一个批处理的示例类。不过官方archetype生成的pom有时候依赖比较全，包含了一些你可能用不到的连接器，我一般会手动精简一下，只留核心依赖，需要什么再加什么。

如果你不想用archetype，也可以手动建一个普通Maven项目，然后自己写pom。手动建的好处是你对每一行依赖都心里有数，坏处是容易漏配置。我两种方式都用过，新手建议先用archetype跑通，再回头精简。

3.2 pom.xml核心依赖：哪些必须，哪些要小心

pom.xml是本地环境搭建的重灾区，配错了就是各种ClassNotFound。我把核心依赖分成三类：Flink核心依赖、日志依赖、测试依赖。

Flink核心依赖里，最基础的是这几个：

<properties> <flink.version>1.17.2</flink.version> <scala.binary.version>2.12</scala.binary.version> <java.version>11</java.version> </properties> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-clients</artifactId> <version>${flink.version}</version> </dependency> </dependencies>

这里有个关键点：flink-clients这个依赖很多人会漏掉。在Flink 1.15之后，客户端相关的类被拆到了单独的模块里，如果你只引flink-streaming-java，运行的时候会报找不到StreamExecutionEnvironment的工厂类。所以这两个要一起引。

日志依赖这块，Flink用的是SLF4J门面，底层实现推荐Log4j2。你需要引：

<dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-slf4j-impl</artifactId> <version>2.20.0</version> </dependency> <dependency> <groupId>org.apache.logging.log4j</groupId> <artifactId>log4j-core</artifactId> <version>2.20.0</version> </dependency>

如果不引日志实现，运行的时候会看到SLF4J的警告，虽然不影响运行，但日志刷不出来，调试的时候很难受。

还有一个坑是依赖范围。在本地IDE运行时，Flink依赖用默认的compile范围就行。但如果你要打fat jar提交到集群，就要把Flink核心依赖设成provided，否则会和集群自带的依赖冲突。这个区别要记清楚，本地开发和打包提交是两套配置。

3.3 日志配置：让控制台输出变得可读

Flink默认的日志配置在conf目录下，但IDE运行时用的是classpath里的log4j2.properties。archetype生成的resources目录下会有一个，内容大概是设置root logger级别和appender。我一般会改成这样：

rootLogger.level = INFO rootLogger.appenderRef.console.ref = ConsoleAppender appender.console.name = ConsoleAppender appender.console.type = CONSOLE appender.console.layout.type = PatternLayout appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss} %-5p %c{1}:%L - %m%n

这个配置的作用是把日志打到控制台，格式包含时间、级别、类名、行号和消息。调试的时候，你能清楚看到任务启动、算子执行、数据输出的过程。如果日志级别设成DEBUG，输出会非常多，新手容易被淹没，所以先用INFO。

注意：如果你在代码里用了System.out.println，输出会和日志混在一起，建议统一用LOG.info，方便过滤。

4. WordCount代码实现与运行验证

4.1 流处理WordCount：从Socket到控制台

WordCount是流处理的“Hello World”，但它麻雀虽小五脏俱全，包含了Source、Transformation、Sink三个部分。我用Socket作为数据源，这样你可以手动输入单词，实时看到统计结果。

先看完整代码：

package com.example.flink; 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 SocketWordCount { public static void main(String[] args) throws Exception { // 1. 创建流处理执行环境 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); // 设置并行度为1，方便本地观察输出顺序 env.setParallelism(1); // 2. 从Socket读取数据 DataStream<String> text = env.socketTextStream("localhost", 9999); // 3. 切分、计数 DataStream<Tuple2<String, Integer>> counts = text .flatMap(new Tokenizer()) .keyBy(value -> value.f0) .sum(1); // 4. 输出到控制台 counts.print(); // 5. 启动任务 env.execute("Socket WordCount"); } public static class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> { @Override public void flatMap(String value, Collector<Tuple2<String, Integer>> out) { String[] words = value.toLowerCase().split("\\s+"); for (String word : words) { if (word.length() > 0) { out.collect(new Tuple2<>(word, 1)); } } } } }

这段代码的逻辑很直白：创建环境、读Socket、切词、按单词分组、求和、打印。但有几个细节值得说。

第一，env.setParallelism(1)这行在本地调试时非常有用。默认并行度是你机器的CPU核数，多个并行子任务同时输出，控制台会交错打印，你根本看不清哪个结果对应哪条输入。设成1之后，输出顺序和输入顺序基本一致，方便验证逻辑。

第二，keyBy这里用的是lambda表达式value -> value.f0，这是Flink 1.12之后推荐的写法。早期版本用keyBy(0)这种索引方式，现在虽然还能用，但会有警告，而且类型不安全。

第三，sum(1)表示对Tuple2的第二个字段求和。Tuple2的字段是f0和f1，索引从0开始，所以1就是第二个字段。

4.2 本地运行步骤：终端、IDE、结果观察

代码写好了，怎么跑起来？步骤其实不复杂，但顺序要对。

第一步，在终端启动一个Socket服务。Linux和Mac用nc命令：

nc -lk 9999

Windows可以用PowerShell或者装一个netcat工具。启动之后，终端会等待输入。

第二步，在IDE里右键运行SocketWordCount的main方法。这时候Flink会在本地启动一个MiniCluster，你会看到日志里打印出JobManager和TaskManager启动的信息。

第三步，回到nc的终端，输入一些单词，比如：

hello world hello flink flink wordcount

每输入一行回车，IDE的控制台就会输出对应的统计结果，类似：

(hello,1) (world,1) (hello,2) (flink,1) (flink,2) (wordcount,1)

这里要注意，print()输出的是Tuple2的toString，格式是(单词, 计数)。因为并行度是1，所以每次更新都会打印一条，你会看到同一个单词的计数不断累加。

提示：如果nc命令在Mac上提示端口被占用，换个端口就行，比如9998，代码里对应改一下。

4.3 批处理WordCount：用DataSet还是DataStream

Flink早期有DataSet API专门做批处理，但从1.12开始，官方主推用DataStream API做批处理，DataSet API在1.18之后已经标记为过时。所以我不建议新手再学DataSet，直接用DataStream的批模式。

批处理WordCount和流处理的主要区别是数据源。流处理用socketTextStream，批处理用readTextFile或者fromElements。比如：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setRuntimeMode(RuntimeExecutionMode.BATCH); DataStream<String> text = env.fromElements( "hello world", "hello flink", "flink wordcount" ); DataStream<Tuple2<String, Integer>> counts = text .flatMap(new Tokenizer()) .keyBy(value -> value.f0) .sum(1); counts.print(); env.execute("Batch WordCount");

关键在env.setRuntimeMode(RuntimeExecutionMode.BATCH)这行，它告诉Flink用批处理模式执行。批模式下，Flink会做更多的优化，比如排序合并，输出结果也更接近传统批处理。

不过说实话，本地测试的时候，流模式和批模式的代码结构几乎一样，区别主要在数据源和执行模式。新手先把流模式跑通，批模式自然就理解了。

5. 常见报错与排查经验实录

5.1 依赖冲突：NoSuchMethodError和ClassNotFoundException

依赖冲突是本地开发最常见的问题，表现通常是运行时报NoSuchMethodError、ClassNotFoundException或者NoClassDefFoundError。这类问题的根源是classpath里有多个版本的同一个类。

排查方法是用Maven的依赖树命令：

mvn dependency:tree -Dincludes=org.apache.flink

这条命令会打印出所有Flink相关的依赖及其版本。如果你看到同一个artifactId出现多个版本，那就是冲突了。解决办法是在pom里用<exclusions>排除掉旧版本，或者用<dependencyManagement>统一版本。

我遇到过一个典型情况：项目里同时引了flink-streaming-java和某个连接器，连接器依赖了旧版Flink，导致运行时加载了旧版的类。这种问题光看代码是看不出来的，必须查依赖树。

5.2 日志不输出：SLF4J绑定缺失

日志不输出，控制台只有Flink的启动信息，没有你的LOG.info内容，大概率是SLF4J没有绑定实现。表现是启动时看到这样的警告：

SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder". SLF4J: Defaulting to no-operation (NOP) logger implementation

解决办法就是前面说的，引log4j-slf4j-impl和log4j-core。注意版本要匹配，log4j-slf4j-impl的版本要和log4j-core一致，否则也可能出问题。

5.3 端口占用与Socket连接失败

SocketWordCount运行时报Connection refused，通常是nc没启动或者端口不对。检查步骤：先在终端确认nc在监听，用netstat -an | grep 9999看端口状态；然后确认代码里的host和port和nc一致。

还有一个坑是nc的-k参数。有些系统的nc不支持-k，导致每次连接只能发一次数据。如果遇到这种情况，可以换用ncat或者写一个简单的Python脚本模拟Socket服务。

5.4 并行度导致的输出乱序

前面提到过，默认并行度是CPU核数，多个子任务同时输出会导致控制台结果交错。除了设setParallelism(1)，还可以在print的时候加前缀，比如counts.print("result")，这样输出会带上子任务标识，方便区分。

下面这张表是我整理的常见问题速查：

提示：遇到报错先看日志的第一行和最后一行，第一行通常是根因，最后一行是触发点，中间是调用栈。不要被长长的调用栈吓到，抓住根因就行。

6. 本地环境进阶：让开发效率再高一点

6.1 用Maven Profile区分本地和打包

本地开发和打包提交的依赖范围不同，用Maven Profile可以一套pom两种配置。比如定义两个profile：

<profiles> <profile> <id>local</id> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> </dependency> </dependencies> </profile> <profile> <id>cluster</id> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java</artifactId> <version>${flink.version}</version> <scope>provided</scope> </dependency> </dependencies> </profile> </profiles>

本地运行时激活local profile，打包时激活cluster profile。这样就不用每次手动改scope了。

6.2 单元测试：用MiniClusterWithClientResource

Flink提供了测试工具类，可以在单元测试里启动一个MiniCluster，不依赖外部环境。核心是MiniClusterWithClientResource，配合JUnit使用。这样你写的算子逻辑可以直接在测试里验证，不用每次都手动跑main方法。

测试依赖需要引：

<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-test-utils</artifactId> <version>${flink.version}</version> <scope>test</scope> </dependency>

然后写测试类，用@ClassRule启动MiniCluster，用env.fromElements构造测试数据，用env.execute触发执行，最后断言结果。这套流程搭好之后，改代码跑测试就是几秒钟的事，比手动跑main快得多。

6.3 调试技巧：断点、日志、Web UI

本地运行Flink任务时，IDE的断点是可以正常打的。你可以在flatMap或者keyBy后面加断点，观察数据流。不过要注意，Flink的算子可能在不同线程执行，断点命中时线程会暂停，如果并行度大于1，其他子任务可能还在跑，观察起来会有点乱。所以调试时还是建议并行度设1。

另外，Flink本地运行时会启动一个Web UI，默认端口8081。你可以在浏览器打开http://localhost:8081，看到任务的执行计划、算子链、数据流量等信息。这个UI对理解Flink的执行模型非常有帮助，建议跑任务的时候顺手打开看看。

日志方面，除了控制台，Flink还会在临时目录生成日志文件。如果你在控制台看不到某些信息，可以去日志文件里翻。日志文件路径通常在启动日志里会打印出来。

7. 我个人在实际操作中的几点体会

搭这套环境，我前前后后在不同机器上重复过很多次，有几个体会比较深。

第一，版本一致性比版本新旧更重要。我见过太多人为了用某个新特性，把Flink升到最新版，结果连接器不兼容，折腾半天又降回来。对于学习和日常开发，选一个稳定的版本，把整套依赖都对齐，比追新有价值得多。

第二，本地环境不要追求和生产一模一样。本地就是用来快速验证逻辑的，分布式、高可用、状态后端这些，等逻辑跑通了再去集群上验证。把本地环境搞得太复杂，反而拖慢开发节奏。

第三，日志和依赖树是两个最值得花时间学的技能。大部分本地开发的问题，都能通过看日志和查依赖树解决。这两个技能学会了，后面遇到问题基本能自己定位，不用到处问人。

第四，WordCount虽然简单，但值得反复写。我每次学一个新引擎，都会先用它写一遍WordCount，因为这个小程序能把数据源、算子、分组、输出这条链路完整走一遍。走通了，心里就有底了。

最后再分享一个小技巧：如果你在本地跑流任务，任务一直不结束是正常的，因为Socket源在持续等待数据。想让它结束，可以手动停止，或者改用批模式，或者用env.fromElements这种有界数据源。别以为是程序卡死了，流任务本来就是一直运行的。

这个环境搭好之后，后面就可以开始玩窗口、状态、时间语义这些更有意思的东西了。下一篇我会在这个基础上，聊聊Flink的窗口机制到底是怎么回事，以及怎么用代码验证不同类型的窗口行为。