1. 从一个最朴素的词频统计说起1.1 为什么词频统计是流处理的第一课但凡接触过大数据处理框架的人几乎都绕不开一个入门案例——词频统计。这东西就像学编程时的Hello World看着简单但里面藏着的东西一点都不少。我刚开始接触Flink那会儿也是从一个WordCount的Demo入手当时觉得不就是数单词嘛能有多复杂结果真正跑起来才发现从环境搭建到任务提交中间踩的坑足够写一篇避雷指南了。WordCount之所以成为流处理框架的经典入门案例原因很实在它同时覆盖了数据读取、数据切分、分组聚合、结果输出这几个核心环节。你把这几个环节跑通了基本上就理解了Flink的DataStream API是怎么运转的。而且词频统计这个场景本身足够直观不需要额外的业务背景知识就能理解输入和输出之间的关系学习成本低反馈却很直接。这篇文章我打算把Flink WordCount这个Demo从里到外讲透包括代码怎么写、每一步为什么这么写、单机环境下怎么把任务提交上去、提交过程中会遇到哪些问题、怎么排查。适合刚接触Flink的开发者也适合之前跑过Demo但没深究细节的朋友。我会尽量用大白话把原理说清楚同时给出可以直接复现的操作步骤。1.2 这个Demo能帮你建立什么认知很多人跑完WordCount就结束了觉得已经学会Flink了。实际上这个Demo能帮你建立三个关键认知第一Flink的数据流模型是怎么回事数据从Source到Transformation再到Sink是怎么流转的第二并行度和算子链的概念为什么你的任务在单机上也能体现出分布式计算的特征第三任务提交的完整链路从写代码到打包到提交到集群每一步在做什么。这三个认知建立起来之后你再去学窗口、状态管理、容错机制就会有一个稳固的基础。反过来如果只是复制粘贴代码跑通了事后面遇到问题还是会一头雾水。所以我的建议是这个Demo值得花时间吃透不要急着往下走。2. 环境准备与项目搭建2.1 版本选择与依赖管理Flink的版本迭代比较快不同版本之间的API有差异。目前主流使用的版本集中在1.13到1.17之间我建议选择1.14或1.15这两个稳定版本社区资料多踩坑成本低。太新的版本可能有些API还在调整太老的版本又缺少一些好用的特性。构建工具用Maven就行Flink官方对Maven的支持最成熟。在pom.xml里需要引入的核心依赖是flink-streaming-java和flink-clients。这里有个细节要注意如果你只是本地跑Demo依赖的scope用默认的compile就行但如果要打包提交到集群Flink的核心依赖应该设为provided因为集群上已经有这些jar包了打进去反而可能导致冲突。properties flink.version1.14.6/flink.version scala.binary.version2.12/scala.binary.version /properties dependencies dependency groupIdorg.apache.flink/groupId artifactIdflink-streaming-java_${scala.binary.version}/artifactId version${flink.version}/version /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-clients_${scala.binary.version}/artifactId version${flink.version}/version /dependency /dependencies注意Flink的artifactId里带有Scala版本号后缀比如_2.12。这是因为Flink部分模块用Scala编写需要匹配对应的Scala二进制版本。如果你用的Scala版本是2.11就要改成_2.11。选错了会在运行时报NoClassDefFoundError。2.2 本地开发环境的最低要求单机跑Flink Demo对硬件要求不高普通开发机就能胜任。JDK版本建议用8或11Flink 1.14对JDK 8的支持最稳定JDK 11也没问题但JDK 17可能会有一些模块化相关的报错。Maven版本用3.6以上即可。IDE方面IntelliJ IDEA是首选它对Maven项目和Scala混编的支持比较好。如果你用Eclipse也不是不行但在依赖管理和代码提示方面会稍微麻烦一些。我个人的经验是IDEA社区版就够用了不需要非得用旗舰版。内存方面本地跑Demo给JVM分配1到2GB就够了。但如果你要开Flink的Web UI建议给到2GB以上不然界面加载可能会卡。这个在后面讲任务提交的时候会再提到。2.3 项目目录结构规划一个清晰的项目结构能让后续的开发和调试省不少事。我通常会把代码分成几个包source放数据源相关的类transform放转换逻辑sink放输出相关的类main放主类入口。对于WordCount这个Demo来说代码量不大但养成好习惯没坏处。flink-wordcount-demo/ ├── pom.xml └── src/ └── main/ ├── java/ │ └── com/ │ └── demo/ │ └── wordcount/ │ ├── WordCount.java │ └── WordCountBatch.java └── resources/ └── log4j2.propertieslog4j2.properties这个文件容易被忽略但很重要。Flink默认会打大量INFO级别的日志如果不配置日志级别控制台会被刷屏调试的时候很难看清自己打印的输出。我一般会把root logger设为WARN只在自己关心的包下开INFO。3. 核心代码逐行拆解3.1 流处理版本的WordCount实现先看完整的流处理版本代码然后我逐段拆解public class WordCount { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); DataStreamString text env.socketTextStream(localhost, 9999); DataStreamTuple2String, Integer counts text .flatMap(new Tokenizer()) .keyBy(value - value.f0) .sum(1); counts.print(); 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)); } } } } }这段代码看着不长但每一行都有讲究。getExecutionEnvironment()这个方法会根据运行上下文自动判断是本地环境还是集群环境写代码的时候不用改部署的时候也不用改这是Flink做得比较优雅的一个地方。setParallelism(1)这行在本地调试时建议加上。因为默认并行度等于CPU核数如果你机器是8核那就会起8个线程打印出来的结果顺序会乱调试的时候看着头疼。设为1之后所有数据走一个线程输出顺序就是确定的。socketTextStream是从Socket端口读取数据这是本地测试最方便的数据源。你开一个终端用nc -lk 9999命令启动一个监听端口然后在里面敲文字Flink就能实时读到。这个方式比读文件更直观因为你能看到数据一条一条进来的效果。3.2 flatMap和keyBy到底在做什么flatMap这个算子名字里的flat指的是扁平化。它的输入是一条记录输出可以是零条、一条或多条记录。在WordCount里输入是一行文本输出是这行文本里每个单词对应的(word, 1)二元组。为什么用flatMap而不是map因为map是一进一出而我们需要一行变多行所以必须用flatMap。keyBy是Flink里最核心的算子之一它做的是逻辑上的分组。注意我说的是逻辑上因为物理上数据可能分布在不同的并行子任务里。keyBy之后相同key的数据会被路由到同一个子任务这样才能保证聚合结果的正确性。在代码里keyBy(value - value.f0)表示按照二元组的第一个字段也就是单词来分组。sum(1)是一个聚合算子表示对第二个字段也就是计数做累加。这里传入的1是字段的位置索引从0开始算所以1代表第二个字段。这个写法在Flink的Tuple类型里很常见但如果你用的是POJO或者自定义类型就需要用lambda表达式指定字段了。3.3 从流处理到批处理的切换Flink有一个比较独特的设计同一套API可以处理流数据和批数据。在1.12版本之前批处理需要用ExecutionEnvironment和DataSetAPI1.12之后官方推荐统一用DataStreamAPI通过设置执行模式来区分流和批。StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment(); env.setRuntimeMode(RuntimeExecutionMode.BATCH); DataStreamString text env.readTextFile(input/words.txt);RuntimeExecutionMode.BATCH模式下Flink会对任务进行优化比如在聚合前先对数据进行排序减少状态的使用。对于有界数据比如读文件批模式通常比流模式快。但如果你要处理的是无界数据比如Socket流就必须用流模式。提示在1.14版本中readTextFile在流模式下默认是逐行读取的读完之后任务不会自动结束。如果你想让任务在读完文件后停止要么用批模式要么在流模式下加一个env.setParallelism(1)并配合DataStream的windowAll来触发有界计算。这个细节很多教程不会讲但实际用的时候一定会遇到。4. 单机任务提交的完整流程4.1 本地执行模式与集群提交的区别Flink的任务执行有两种基本方式本地执行和集群提交。本地执行就是在IDE里直接run main方法Flink会在当前JVM里启动一个MiniCluster所有组件都跑在一个进程里。这种方式适合开发和调试但不适合生产。集群提交是把打好的jar包提交到一个独立运行的Flink集群上。集群可以是一个Standalone集群也可以是on YARN或on Kubernetes的集群。单机环境下我们通常搭一个Standalone集群来模拟。两者的核心区别在于本地执行时getExecutionEnvironment()返回的是一个LocalStreamEnvironment集群提交时返回的是一个RemoteStreamEnvironment。代码本身不需要改Flink会自动判断。但集群提交时有些依赖需要额外注意比如连接器相关的jar包需要一起打进去。4.2 搭建单机Standalone集群单机Standalone集群的搭建步骤不复杂但有几个配置项容易出错。先去官网下载Flink的二进制包解压到某个目录然后修改conf/flink-conf.yaml里的几个关键配置。jobmanager.rpc.address: localhost jobmanager.rpc.port: 6123 jobmanager.memory.process.size: 1024m taskmanager.memory.process.size: 1024m taskmanager.numberOfTaskSlots: 4 parallelism.default: 2jobmanager.rpc.address设为localhost因为单机环境下JobManager和TaskManager在同一台机器上。taskmanager.numberOfTaskSlots表示每个TaskManager能提供多少个任务槽这个值决定了集群的最大并行度。单机测试设为4就够了设太大反而会占用过多内存。内存配置要注意jobmanager.memory.process.size和taskmanager.memory.process.size是进程总内存包括JVM堆内存、堆外内存、元空间等。如果设得太小启动时可能会报内存不足的错误。我建议单机测试时每个进程至少给1024MB如果机器内存充裕给2048MB更稳妥。配置改完之后用bin/start-cluster.sh启动集群。启动成功后访问http://localhost:8081应该能看到Flink的Web UI。如果打不开先检查防火墙设置再检查日志文件log/flink-*-jobmanager-*.log里有没有报错。4.3 打包与提交任务代码写完之后用Maven打包mvn clean package -DskipTests打包完成后在target目录下会生成一个jar文件。提交任务的命令是bin/flink run -c com.demo.wordcount.WordCount \ flink-wordcount-demo-1.0-SNAPSHOT.jar-c参数指定主类的全限定名。如果你的jar包里只有一个主类可以省略这个参数Flink会自动找到。但如果有多个主类就必须指定否则会报Could not find the main class的错误。提交成功后Web UI的Running Jobs页面会出现你的任务。点击进去可以看到任务的执行计划图包括Source、flatMap、keyBy、sum、Sink这几个算子以及它们之间的数据流转关系。这个图对于理解Flink的任务拓扑非常有帮助。注意如果你在代码里用了Socket作为数据源提交到集群后需要确保Socket服务在运行并且集群所在的机器能访问到那个端口。如果Socket服务在宿主机上而Flink跑在容器里可能会因为网络隔离导致连不上。这种情况下要么把Socket服务也放到容器里要么用文件作为数据源。4.4 任务提交后的验证与日志查看任务提交之后怎么确认它真的在跑最直接的方式是看Web UI上的指标。在任务详情页可以看到接收的记录数、发送的记录数、各算子的吞吐量等指标。如果这些数字在变化说明任务在正常处理数据。另一个方式是看日志。TaskManager的日志在log/flink-*-taskmanager-*.out文件里里面会打印print()算子的输出。如果你在代码里用了counts.print()那么每个单词的计数结果会出现在这个日志里。注意print()算子默认是打印到标准输出在集群模式下会被重定向到TaskManager的日志文件。如果任务提交后立刻失败先看JobManager的日志通常会有异常堆栈。常见的错误包括主类找不到、依赖冲突、端口被占用、内存不足。这些问题的排查方法我会在下一节详细讲。5. 常见问题与排查技巧实录5.1 任务提交失败的典型原因我整理了一个速查表覆盖了单机提交WordCount任务时最常遇到的几类问题问题现象可能原因排查方法解决方案提交时报ClassNotFoundException主类名写错或jar包不完整用jar tf查看jar包内容检查-c参数确认主类全限定名正确任务启动后立即失败依赖冲突或内存不足查看JobManager日志的异常堆栈排除冲突依赖调大内存配置Socket连接被拒绝Socket服务未启动或端口不通用telnet测试端口连通性启动Socket服务检查防火墙任务一直处于RUNNING但无输出数据源没有数据流入检查Socket端是否输入了内容在Socket终端输入测试文本Web UI打不开端口被占用或防火墙拦截用netstat查看8081端口修改rest.port配置或关闭防火墙打包后jar包过大把Flink核心依赖打进去了查看jar包大小和内容将Flink依赖的scope设为provided这个表里的每一行都是我实际踩过的坑。特别是最后一条我第一次打包的时候打出来的jar有80多MB提交上去各种冲突。后来把Flink核心依赖设为providedjar包一下子降到几MB问题就没了。5.2 并行度设置不当引发的诡异现象并行度是Flink里最容易让人困惑的概念之一。我见过不少人在本地跑WordCount时发现输出结果里同一个单词出现了多次每次计数都是1而不是累加。这个现象的根本原因就是并行度大于1而keyBy之后的聚合是在多个子任务里分别进行的。举个例子假设并行度是2数据里的hello这个单词被分到了子任务1world被分到了子任务2。那么子任务1会输出(hello, 1)子任务2会输出(world, 1)。如果你期望看到(hello, 5)这样的累加结果就必须保证相同key的数据进入同一个子任务。keyBy做的就是这件事但前提是并行度设置要合理。在本地调试时最简单的做法就是把并行度设为1这样所有数据走一个线程聚合结果一定是正确的。但如果你要测试并行场景下的行为就需要理解keyBy之后的数据分区逻辑。Flink默认使用哈希分区相同key的哈希值相同所以会被分到同一个子任务。这个机制在大多数情况下是可靠的但如果你自定义了key的选择逻辑就要确保hashCode方法的实现是正确的。5.3 日志配置与调试输出技巧Flink默认的日志级别是INFO控制台会输出大量框架内部的日志。调试的时候这些日志会淹没你自己的输出。我通常会在log4j2.properties里做如下配置rootLogger.level WARN rootLogger.appenderRef.console.ref ConsoleAppender logger.wordcount.name com.demo.wordcount logger.wordcount.level INFO logger.wordcount.appenderRef.console.ref ConsoleAppender这样配置之后只有com.demo.wordcount包下的日志会以INFO级别输出其他框架日志只显示WARN及以上级别。控制台一下子就清爽了。另外print()算子虽然方便但在生产环境不要用因为它会把结果直接打到标准输出没有持久化也没有背压处理。生产环境应该用addSink()把结果写到外部存储比如文件、数据库或消息队列。对于Demo来说print()够用了但要知道它的局限性。5.4 单机资源不足时的降级方案单机跑Flink集群最容易遇到的就是内存不足。特别是当你同时开着IDE、浏览器、Flink集群的时候8GB内存的机器可能会比较吃力。如果遇到OutOfMemoryError或者集群启动后TaskManager频繁重启可以考虑以下几个降级方案。第一降低TaskManager的内存配置。taskmanager.memory.process.size可以从1024m降到768m但不要再低了再低可能连JVM都启动不了。第二减少TaskManager的数量。单机环境下默认启动1个TaskManager就够了不需要多个。第三关闭不必要的Flink组件比如历史服务器History Server这个组件会额外占用内存。如果实在跑不动集群那就退回本地执行模式。在IDE里直接run main方法Flink会启动一个MiniCluster资源占用比完整集群小得多。虽然不能完全模拟集群环境但对于验证代码逻辑来说已经足够了。6. 从Demo到实战的扩展思路6.1 数据源和输出端的替换WordCount这个Demo用的是Socket作为数据源print()作为输出。实际项目中数据源可能是Kafka、文件系统、数据库输出端可能是Elasticsearch、MySQL、HDFS。替换数据源和输出端的时候核心逻辑flatMap、keyBy、sum不需要改只需要换掉Source和Sink的代码。以Kafka为例你需要引入flink-connector-kafka依赖然后把socketTextStream换成KafkaSource。KafkaSource的配置包括broker地址、topic名称、消费者组ID、反序列化器等。这些配置项在官方文档里都有说明但有一个细节要注意Kafka连接器的版本必须和Flink版本匹配否则会出现兼容性问题。输出端换成MySQL的话可以用JdbcSink。需要提供一个JdbcStatementBuilder来构造INSERT语句以及一个JdbcConnectionOptions来配置数据库连接。这里有个性能上的考量如果每条记录都单独插入数据库压力会很大。通常的做法是攒一批再插入或者用连接池来复用连接。6.2 状态管理与容错机制的引入WordCount这个Demo没有用到Flink的状态管理因为它的聚合逻辑是无状态的——每个单词的计数只依赖于当前记录不依赖于历史记录。但如果你要做滑动窗口统计比如过去5分钟内每个单词出现的次数就需要用到状态了。Flink的状态分为Keyed State和Operator State。Keyed State是和key绑定的只能在keyBy之后的算子中使用。对于WordCount的扩展场景你可以用ValueState来存储每个单词的累计计数用ListState来存储窗口内的记录。状态的后端可以选择MemoryStateBackend适合本地调试、FsStateBackend适合生产环境或RocksDBStateBackend适合超大状态。容错机制的核心是Checkpoint。Flink会定期对状态做快照当任务失败时从最近的Checkpoint恢复。Checkpoint的配置包括间隔时间、超时时间、最大并发数等。间隔太短会影响性能太长会导致恢复时丢失较多数据。通常建议设为1到5分钟具体取决于业务对数据丢失的容忍度。6.3 性能调优的几个入手点当WordCount的数据量从几条变成几百万条时性能问题就会暴露出来。调优的入手点主要有三个并行度、算子链、缓冲区。并行度的调整比较直接增加并行度可以提升吞吐量但也会增加资源消耗和协调开销。通常的做法是先测出单并行度的吞吐量然后根据目标吞吐量计算需要的并行度。但要注意并行度不是越高越好当并行度超过数据分区的数量时多余的子任务会空转。算子链是Flink的一个优化机制它会把满足条件的相邻算子合并成一个任务减少线程间的切换和数据序列化开销。但有时候算子链反而会降低性能比如当某个算子的处理速度明显慢于上游时合并在一起会导致上游被阻塞。这种情况下可以用disableChaining()或startNewChain()来打断算子链。缓冲区的大小影响网络传输的效率。缓冲区太小会导致频繁的网络往返太大则会增加延迟。Flink提供了taskmanager.memory.network.fraction等配置项来调整网络缓冲区的内存占比。默认值在大多数场景下是够用的但如果你的数据记录特别大或者特别小可能需要手动调整。6.4 监控指标的关注重点任务跑起来之后怎么知道它健不健康Web UI上有一堆指标但不需要每个都看。我通常关注这几个numRecordsIn和numRecordsOut看数据吞吐量currentSendTime和currentFetchTime看网络延迟checkpointDuration和checkpointSize看状态快照的开销backPressure看是否有背压。背压是流处理里最需要警惕的信号。它表示下游处理速度跟不上上游数据在缓冲区里堆积。短时间的背压是正常的但如果持续存在就说明某个算子成了瓶颈。定位背压的方法是在Web UI的BackPressure页面查看各个算子的背压状态红色的表示背压严重需要重点排查。排查背压的思路通常是先看是不是数据倾斜导致的某个key的数据量远大于其他key再看是不是某个算子的处理逻辑太重比如里面有同步的IO操作最后看资源是否充足CPU、内存、网络带宽有没有到瓶颈。定位到原因之后对应的解决方案包括加盐打散热点key、异步化IO操作、增加并行度或资源。7. 一些个人体会Flink的WordCount Demo我前前后后跑过不下十次每次都有新的收获。最开始只是照着教程敲代码能跑通就满足了。后来开始琢磨每一步为什么这么写为什么要用flatMap而不是map为什么要keyBy并行度设多少合适。再后来开始折腾集群提交踩了一堆坑才慢慢理解Flink的部署模型和资源管理机制。如果你刚开始学Flink我的建议是不要急着跳过这个Demo。花点时间把代码吃透把集群搭起来把任务提交上去把遇到的问题一个一个解决掉。这个过程积累下来的经验比看十篇教程都管用。而且WordCount虽然简单但它涉及的核心概念——数据流、算子、并行度、状态、Checkpoint——都是Flink的基石这些概念理解透了后面学什么都快。最后分享一个小技巧如果你在本地调试时发现输出结果不符合预期先检查并行度是不是大于1。这个坑我踩过好几次每次都是因为忘了设setParallelism(1)导致聚合结果被分散到多个子任务里。设成1之后问题通常就消失了。如果设成1还是不对那就检查keyBy的字段选择是否正确以及数据里有没有空字符串或特殊字符干扰了分词逻辑。