★ FEATURED ARTICLE 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的反射机制会报错。下面这张表是我总结的版本搭配建议可以直接参考Flink版本推荐Scala版本推荐Java版本适用场景1.17.x2.12Java 8 / 11稳定学习、连接器丰富1.18.x2.12Java 8 / 11 / 17新特性尝鲜、长期使用1.19.x2.12Java 11 / 17跟进新版本1.20.x2.12Java 11 / 17最新特性验证提示不要同时在一个项目里混用多个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 \ -DarchetypeGroupIdorg.apache.flink \ -DarchetypeArtifactIdflink-quickstart-java \ -DarchetypeVersion1.17.2 \ -DgroupIdcom.example.flink \ -DartifactIdflink-wordcount-demo \ -Dversion1.0-SNAPSHOT \ -Dpackagecom.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.version1.17.2/flink.version scala.binary.version2.12/scala.binary.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-clients这个依赖很多人会漏掉。在Flink 1.15之后客户端相关的类被拆到了单独的模块里如果你只引flink-streaming-java运行的时候会报找不到StreamExecutionEnvironment的工厂类。所以这两个要一起引。日志依赖这块Flink用的是SLF4J门面底层实现推荐Log4j2。你需要引dependency groupIdorg.apache.logging.log4j/groupId artifactIdlog4j-slf4j-impl/artifactId version2.20.0/version /dependency dependency groupIdorg.apache.logging.log4j/groupId artifactIdlog4j-core/artifactId version2.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读取数据 DataStreamString text env.socketTextStream(localhost, 9999); // 3. 切分、计数 DataStreamTuple2String, 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 FlatMapFunctionString, Tuple2String, Integer { Override public void flatMap(String value, CollectorTuple2String, 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 9999Windows可以用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还是DataStreamFlink早期有DataSet API专门做批处理但从1.12开始官方主推用DataStream API做批处理DataSet API在1.18之后已经标记为过时。所以我不建议新手再学DataSet直接用DataStream的批模式。批处理WordCount和流处理的主要区别是数据源。流处理用socketTextStream批处理用readTextFile或者fromElements。比如StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setRuntimeMode(RuntimeExecutionMode.BATCH); DataStreamString text env.fromElements( hello world, hello flink, flink wordcount ); DataStreamTuple2String, 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 -Dincludesorg.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)这样输出会带上子任务标识方便区分。下面这张表是我整理的常见问题速查现象可能原因排查方法解决方式NoSuchMethodError依赖版本冲突mvn dependency:tree排除旧版本或统一版本ClassNotFoundException缺少依赖检查pom补充对应依赖日志不输出SLF4J无绑定看启动警告引log4j-slf4j-implConnection refusedSocket未启动netstat查端口启动nc或换端口输出乱序并行度大于1观察输出setParallelism(1)任务不结束流任务持续运行正常现象手动停止或改批模式提示遇到报错先看日志的第一行和最后一行第一行通常是根因最后一行是触发点中间是调用栈。不要被长长的调用栈吓到抓住根因就行。6. 本地环境进阶让开发效率再高一点6.1 用Maven Profile区分本地和打包本地开发和打包提交的依赖范围不同用Maven Profile可以一套pom两种配置。比如定义两个profileprofiles profile idlocal/id dependencies dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink.version}/version /dependency /dependencies /profile profile idcluster/id dependencies dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java/artifactId version${flink.version}/version scopeprovided/scope /dependency /dependencies /profile /profiles本地运行时激活local profile打包时激活cluster profile。这样就不用每次手动改scope了。6.2 单元测试用MiniClusterWithClientResourceFlink提供了测试工具类可以在单元测试里启动一个MiniCluster不依赖外部环境。核心是MiniClusterWithClientResource配合JUnit使用。这样你写的算子逻辑可以直接在测试里验证不用每次都手动跑main方法。测试依赖需要引dependency groupIdorg.apache.flink/groupId artifactIdflink-test-utils/artifactId version${flink.version}/version scopetest/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的窗口机制到底是怎么回事以及怎么用代码验证不同类型的窗口行为。
阅读完成 · 觉得有帮助？
