企业官网建设流程全解析

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.version>1.14.6</flink.version> <scala.binary.version>2.12</scala.binary.version> </properties> <dependencies> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-streaming-java_${scala.binary.version}</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-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或11，Flink 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.properties

log4j2.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); DataStream<String> text = env.socketTextStream("localhost", 9999); DataStream<Tuple2<String, Integer>> counts = text .flatMap(new Tokenizer()) .keyBy(value -> value.f0) .sum(1); counts.print(); 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)); } } } } }

这段代码看着不长，但每一行都有讲究。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和DataSetAPI；1.12之后，官方推荐统一用DataStreamAPI，通过设置执行模式来区分流和批。

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setRuntimeMode(RuntimeExecutionMode.BATCH); DataStream<String> 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: 2

jobmanager.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任务时最常遇到的几类问题：

这个表里的每一行都是我实际踩过的坑。特别是最后一条，我第一次打包的时候，打出来的jar有80多MB，提交上去各种冲突。后来把Flink核心依赖设为provided，jar包一下子降到几MB，问题就没了。

5.2 并行度设置不当引发的诡异现象

并行度是Flink里最容易让人困惑的概念之一。我见过不少人在本地跑WordCount时，发现输出结果里同一个单词出现了多次，每次计数都是1，而不是累加。这个现象的根本原因就是并行度大于1，而keyBy之后的聚合是在多个子任务里分别进行的。

举个例子，假设并行度是2，数据里的"hello"这个单词被分到了子任务1，"world"被分到了子任务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的字段选择是否正确，以及数据里有没有空字符串或特殊字符干扰了分词逻辑。