一、先把并行度这件事说明白
很多人刚接触Flink的时候，一听到“并行度”三个字就紧张，总觉得这是个高深莫测的概念。其实你完全可以把它想象成食堂打饭。一个窗口打饭，一百个人排队，速度肯定快不了。如果开十个窗口，每个人排队的时间就大大缩短。这里的“窗口数量”就是并行度。
在Flink的世界里，并行度就是同一个算子（比如map、filter、keyBy）同时被多少个线程来执行。并行度越高，同时干活的线程越多，处理数据的能力理论上就越强。但要注意，“理论上”这三个字很关键，因为并行度不是越高越好，食堂开一百个窗口但厨师只有两个，照样快不起来。后面我们会详细聊怎么设置才合理。
并行度的本质，是让数据分散到多个“通道”里，每个通道独立处理一部分数据。这些通道在Flink内部叫“子任务”。一个算子的并行度是3，意思就是这个算子有3个子任务在同时跑。数据源发下来的数据，会被分配到不同的子任务上，每个子任务只处理属于自己那一份。
看到这里你就明白了，并行度直接决定了Flink处理和吞吐数据的能力，也影响着资源消耗。所以搞清楚并行度是怎么来的、怎么调的、怎么优化，是每个用Flink的人绕不开的一课。
二、并行度在Flink里是怎么定下来的
Flink里并行度的设置不是只有一处，它有一套自己的“继承”规则。简单理解，就是“谁离得近谁说了算”。你可以在多个地方设置并行度，Flink会按照优先级取一个最终的值。这个优先级从高到低大致是这样的：
第一，算子自己单独指定。第二，代码里执行环境指定的全局并行度。第三，提交作业时通过命令行参数指定的并行度。第四，配置文件里默认的并行度。
也就是说，如果你在某个算子后面写了 setParallelism(5)，那这个算子死活都是5，不管别处怎么设。如果你没在算子上面设，那就看执行环境有没有设全局并行度。执行环境也没设，那就看提交作业的时候有没有传 -p。都没有，就取配置文件里的默认值。
2.1 代码里指定并行度
最常见的做法，是在Flink程序的执行环境上设置全局并行度。下面这段代码演示了怎么设置。
// 技术栈：Java + Flink 1.17
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class GlobalParallelismExample {
public static void main(String[] args) throws Exception {
// 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 设置整个作业的全局并行度为4
// 如果没有其他更高优先级的设置，所有算子都会默认并行度为4
env.setParallelism(4);
// 这里读的是一个本地文件
// 这个source的并行度也会继承全局的4（除非文件源自身有限制）
env.readTextFile("hdfs:///logs/access.log")
.map(line -> { // 这个map算子也是并行度4
String[] fields = line.split(",");
// 简单处理一行数据，返回用户id
return fields[0];
})
.print(); // 打印到标准输出，并行度也是4
// 执行作业
env.execute("GlobalParallelismExample");
}
}
上面的代码很简单，所有算子都共享同一个并行度。但实际生产里，不同算子的负载差异很大，我们往往需要给不同算子设置不同的并行度。
2.2 给算子单独设置并行度
比如一个数据过滤算子，逻辑非常简单，并行度可以设得高一点；而一个需要访问外部数据库的算子，并发太高反而会把数据库打挂了，这时候就要调低并行度。
// 技术栈：Java + Flink 1.17
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
public class OperatorParallelismExample {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 全局并行度默认设为2
env.setParallelism(2);
// 给source算子单独设置并行度为1
// 例如从消息队列里读数据，只开一个连接就够用了
env.addSource(new RichSourceFunction<String>() {
@Override
public void run(SourceContext<String> ctx) throws Exception {
// 模拟不断产生数据
while (true) {
ctx.collect("{\"userId\":\"1001\",\"action\":\"click\"}");
Thread.sleep(1000);
}
}
@Override
public void cancel() {
// 取消时的清理逻辑
}
}).setParallelism(1) // source并行度是1
// 过滤掉空数据，这个算子逻辑轻，并行度给8
.filter(line -> line != null && !line.isEmpty()).setParallelism(8)
// 解析JSON字符串，提取userId，并行度给4
.map(json -> {
// 假设这里用了一个fastjson工具类
return com.alibaba.fastjson.JSON.parseObject(json).getString("userId");
}).setParallelism(4);
env.execute("OperatorParallelismExample");
}
}
你看到了，每个算子后面都可以加一个 setParallelism()，非常灵活。生产环境里我们经常这样干，让轻量算子多线程跑，让重量算子控制好并发。
2.3 提交作业时指定并行度
如果你的代码里既没有设置全局并行度，也没有给算子单独设置，那就可以在提交Flink作业时指定。这个方式适合运维同事在发布时根据自己的集群规模决定。
# 这是一个示例命令，用来演示提交作业时指定并行度
# 技术栈：Java（命令是shell，但这里仅作注释展示）
# 注意-p参数后面跟的是并行度数值
flink run -p 5 -c com.example.MyFlinkJob myflink-job.jar
注意，上面的 -p 5 会让作业的全局并行度变为5，但代码里如果某个算子显式写了 setParallelism(8)，那这个算子还是8，因为算子级别的优先级更高。
2.4 配置文件里的默认并行度
如果上面几种方式都没指定，那就看Flink的配置文件 flink-conf.yaml。里面有一项叫 parallelism.default，默认是1。对于生产环境，我们一般会把它调大一些，比如4或者8。
# flink-conf.yaml 文件内容示例
# 技术栈：YAML 配置（为了保持统一性，这里用注释说明效果）
parallelism.default: 4
不过在实际项目中，依赖这个默认值是很危险的一件事。因为不同的作业负载不一样，写死在代码里，或者提交时用 -p 指定，才是最可控的。
三、影响并行度设置的几个关键东西
知道了并行度怎么设置，接下来要搞明白什么因素会影响你选多大。这里提几个最常见的影响因素。
3.1 数据源的“脾气”
不是所有数据源都支持任意并行度。有的数据源天生只有一个，比如监听某个TCP端口的数据源，并行度只能是1。有的数据源支持并行读取，比如Kafka，一个topic有10个分区，你就算把source的并行度设成20，它也最多同时开10个连接去读，剩下10个是空闲状态。
所以设置source并行度之前，先了解你的数据源支不支持多个并行度。Kafka这种分布式消息队列，并行度一般直接跟分区数挂钩。
3.2 KeyBy之后的分配逻辑
如果你用了 keyBy() 做分组，那么并行度怎么设就更有讲究了。keyBy 会把相同key的数据分到同一个子任务里，这个分配是哈希取模。如果你有100万个不同的key，并行度是10，那每个子任务大约处理10万个key。但如果key分布不均匀，比如有个超级大key的数据量占了80%，那这个key被分到的那个子任务就会特别累，其他子任务却闲着，这就是数据倾斜。
// 技术栈：Java + Flink 1.17
// 演示keyBy后不同并行度的数据分布情况
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class KeyByParallelismExample {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(4);
DataStream<String> stream = env.socketTextStream("192.168.1.10", 9999);
stream
.map(line -> {
String[] arr = line.split(",");
// 返回一个Pair封装：省份,金额
return new Pair(arr[0], Double.parseDouble(arr[1]));
})
// 按照省份分组，同一个省份的所有金额会汇总到同一个并行子任务里
// 如果某个省份的数据特别多，就会造成倾斜
.keyBy((KeySelector<Pair, String>) pair -> pair.province)
.reduce((v1, v2) -> new Pair(v1.province, v1.amount + v2.amount))
.print();
env.execute("KeyByParallelismExample");
}
// 简单的封装类
public static class Pair {
public String province;
public Double amount;
public Pair(String province, Double amount) {
this.province = province;
this.amount = amount;
}
}
}
出现这种倾斜问题的时候，盲目调高并行度是没用的，因为大key还是落在同一个子任务上。你需要借助 rebalance 或者把key加盐（二次分组），让数据更均匀地分散。
3.3 状态的大小
如果算子使用了状态（比如Keyed State），那么并行度变化时，状态数据会根据key重新分布。Flink的Checkpoint会帮你自动做这件事，但也会增加恢复时间。一个状态特别大的算子，并行度调太高，状态分布在几百个子任务里，每个子任务都要从远程存储拉取自己的状态，这个过程可能很耗时。所以生产里不要频繁调整有状态算子的并行度。
四、生产环境里并行度怎么定才靠谱
这块是很多同学最关心的。光知道API怎么用还不够，真正在集群上跑，需要考虑的东西很多。我总结几个实用原则。
4.1 先看资源再定并行度
你的Flink集群有多少CPU、多少内存，是定并行度的前提。比如集群有10台TaskManager，每个TaskManager有4个slot，那整个集群最多同时运行40个任务槽位。你一个作业的并行度加起来，不能超过slot总数。否则作业根本启动不了。
所以一个不严谨但常用的估算方式是：并行度总大小 = 集群总slot数 * 资源预留比例。比如留20%给系统占用，那最大并行度总数就是40 * 0.8 = 32。
4.2 让source并行度跟着分区走
如果是读Kafka，source的并行度最好等于Kafka topic的分区数。你既能保证一个分区对应一个读线程，没有闲置线程，也能保证每个分区被均匀处理。Kafka分区数一般比节点数多，但也不会太夸张。
4.3 计算密集的算子可以调高
比如数据清洗、格式转换、加字段这类纯CPU计算，没有网络等待，没有锁竞争，并行度可以适当调高。因为这类算子只是拼命算，多几个线程就能多摊一部分数据。一般可以设置为CPU核数的2到3倍，但最好不要超过总slot数。
4.4 涉及外部IO的算子要克制
如果你的map算子里要去查MySQL、Redis、或者调用外部HTTP接口，这时候并行度就不能拍脑袋设高。并行度太高，你的外部系统可能瞬间被打爆，比如数据库连接数耗尽、HTTP服务超时。合理做法是先用小并行度压测，观察外部系统的延迟和负载，然后慢慢往上加。
4.5 检查点背压和容错
并行度调高之后，checkpoint的规模也会变大。因为每个子任务都要将自己的状态快照上传到远程存储。如果并行度太高，同一时刻上传大量快照，可能给网络和存储带来压力。所以调并行度时，也要关注checkpoint的时长和失败率。
背压是另一个信号。如果你在Web UI上发现某个算子有严重的背压，说明它的下游处理不过来了。这时候不是盲目提高这个算子的并行度，而是要找到瓶颈算子是哪个。如果是上游发给它的数据量太大，可以增加它的并行度。如果是它自己要查外部数据库，瓶颈在IO，那就要优化查询逻辑，或者加缓存。
五、一个综合的生产调优示例
下面我们结合一个“实时用户行为统计”的场景，把并行度设置的思路串起来。场景是：从Kafka消费用户点击日志，过滤掉无效数据，按用户ID分组，统计每个用户每小时的点击次数，最后写入MySQL。
技术栈统一使用Java。代码里每一步都标注了并行度设置的理由，你可以看到生产环境是怎么灵活配置的。
// 技术栈：Java + Flink 1.17
// 场景：实时统计每个用户每小时的点击数，结果写入MySQL
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.time.Duration;
public class ClickCountJob {
public static void main(String[] args) throws Exception {
// 1. 创建执行环境，全局并行度设置成8
// 这个值是根据集群总slot数和预留资源估算出来的
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(8);
// 2. 从Kafka读取点击日志
// Kafka这个topic有12个分区，所以source并行度直接设成12，跟分区数保持一致
KafkaSource<String> source = KafkaSource.<String>builder()
.setBootstrapServers("192.168.1.20:9092")
.setTopics("user_click_log")
.setGroupId("click_count_group")
.setStartingOffsets(OffsetsInitializer.earliest())
.setValueOnlyDeserializer(new SimpleStringSchema())
.build();
DataStream<String> kafkaStream = env
.fromSource(source, WatermarkStrategy.noWatermarks(), "ClickLogSource")
.setParallelism(12); // 等于Kafka分区数
// 3. 解析日志，过滤无效数据
// 这个阶段是纯CPU计算，没有IO阻塞，可以把并行度调高到16
DataStream<ClickRecord> parsedStream = kafkaStream
.map(new MapFunction<String, ClickRecord>() {
@Override
public ClickRecord map(String line) throws Exception {
String[] fields = line.split("\\|");
if (fields.length < 3) {
// 数据处理不了，返回null，后面过滤掉
return null;
}
long userId = Long.parseLong(fields[0]);
long timestamp = Long.parseLong(fields[1]);
String page = fields[2];
return new ClickRecord(userId, timestamp, page);
}
})
.filter(record -> record != null)
.setParallelism(16); // 此处并行度高于全局，因为计算简单且数据量大
// 4. 按用户ID分组，开1小时滚动窗口，计数
// 这里用了keyBy，并行度设置成12，既能充分利用资源，又不至于让窗口状态太碎片化
DataStream<UserClickCount> countedStream = parsedStream
.assignTimestampsAndWatermarks(
WatermarkStrategy
.<ClickRecord>forBoundedOutOfOrderness(Duration.ofSeconds(10))
.withTimestampAssigner((record, ts) -> record.timestamp)
)
.keyBy(record -> record.userId)
.window(TumblingEventTimeWindows.of(Time.hours(1)))
.aggregate(new CountAggregate(), new WindowResultFunction())
.setParallelism(12); // keyBy后的并行度要根据key数量和数据倾斜情况来定
// 5. 把结果写入MySQL
// 外部数据库的连接数有限，并行度不能太高，设成4
countedStream.addSink(new MysqlSink()).setParallelism(4);
env.execute("ClickCountJob");
}
// ---- 以下是一些辅助类和函数 ----
public static class ClickRecord {
public long userId;
public long timestamp;
public String page;
public ClickRecord(long userId, long timestamp, String page) {
this.userId = userId;
this.timestamp = timestamp;
this.page = page;
}
}
public static class UserClickCount {
public long userId;
public long windowEnd;
public long count;
public UserClickCount(long userId, long windowEnd, long count) {
this.userId = userId;
this.windowEnd = windowEnd;
this.count = count;
}
}
// 预聚合，利用窗口内的增量计算减少状态量
public static class CountAggregate implements org.apache.flink.api.common.functions.AggregateFunction<ClickRecord, Long, Long> {
@Override
public Long createAccumulator() {
return 0L;
}
@Override
public Long add(ClickRecord value, Long accumulator) {
return accumulator + 1L;
}
@Override
public Long getResult(Long accumulator) {
return accumulator;
}
@Override
public Long merge(Long a, Long b) {
return a + b;
}
}
// 从WindowedStream中产出窗口结果
public static class WindowResultFunction
implements org.apache.flink.streaming.api.functions.windowing.WindowFunction<Long, UserClickCount, Long, org.apache.flink.streaming.api.windowing.windows.TimeWindow> {
@Override
public void apply(Long key,
org.apache.flink.streaming.api.windowing.windows.TimeWindow window,
Iterable<Long> input,
org.apache.flink.util.Collector<UserClickCount> out) throws Exception {
Long count = input.iterator().next();
out.collect(new UserClickCount(key, window.getEnd(), count));
}
}
// 自定义MySQL Sink，注意资源释放
public static class MysqlSink extends RichSinkFunction<UserClickCount> {
private Connection conn;
private PreparedStatement stmt;
@Override
public void open(Configuration parameters) throws Exception {
Class.forName("com.mysql.cj.jdbc.Driver");
conn = DriverManager.getConnection(
"jdbc:mysql://localhost:3306/test?useSSL=false",
"root", "password");
stmt = conn.prepareStatement(
"INSERT INTO user_click_count (user_id, window_end, count) VALUES (?, ?, ?) " +
"ON DUPLICATE KEY UPDATE count = VALUES(count)");
}
@Override
public void invoke(UserClickCount value, Context context) throws Exception {
stmt.setLong(1, value.userId);
stmt.setLong(2, value.windowEnd);
stmt.setLong(3, value.count);
stmt.executeUpdate();
}
@Override
public void close() throws Exception {
if (stmt != null) stmt.close();
if (conn != null) conn.close();
}
}
}
上面这个例子就是一套很典型的并行度配置思路：source跟随Kafka分区数，纯计算算子高并行度，keyBy算子适中，Sink高并发低。实际生产里你还需要配合压测和监控来微调，但方向就是这样。
六、什么时候该调整并行度，什么时候不该动
6.1 适合调高并行度的情况
数据量大，但是处理逻辑简单（比如过滤、转换）。集群资源充足，CPU和内存还有很多余量。上游系统支持并发读取，比如Kafka分区数多。任务里没有大量的外部依赖。
6.2 不适合调高并行度的情况
外部系统接口慢，容易被打垮。状态特别大，调整并行度会触发大规模状态重分配。全局并行度已经接近集群slot上限，再提也没意义。你的数据源本身并行度就是1，比如socket，source只有1，后面并行度再高，数据进来也是先排队。
6.3 并行度太高有哪些坑
很多新手一上来就把并行度设成64、128，觉得越大越快。但并行度太高，会带来几个问题：
一是线程上下文切换开销变大。每个线程只处理一小撮数据，大部分时间都在抢CPU。二是网络shuffle增多。数据在不同子任务之间传输需要序列化和网络IO，并行度越高，节点之间的数据流动越频繁。三是状态和checkpoint的负担变重。每个子任务都要做快照，数量多了，上传快照的时间也会增长。四是内存占用变大。每个子任务都有自己的一些缓冲区，并行度太高的话，缓冲区加起来可能吃掉大量内存，甚至导致内存溢出。
6.4 值得注意的几个小细节
并行度不仅和吞吐相关，还和事件时间的watermark推进有关。如果并行度太高，每个子任务都需要配合watermark对齐，可能会加大乱序程度。
rescale 和 rebalance 也能重新分配数据。rebalance 是轮询，每个子任务轮流拿数据，适合负载均衡。rescale 是本地就近分配，减少网络开销。当数据倾斜不明显的时候，用 rescale 通常更好。
七、生产环境并行度优化完整流程
最后给大家总结一套可以落地执行的流程，每一步都有依据，不靠猜。
第一步，看集群资源。确认现有TaskManager数量、每个节点的CPU核数、内存大小、slot数量。算清楚你能用多少个并行slot。
第二步，看数据流拓扑。把Flink作业的整个数据流图画出来，标出哪些算子有状态、哪些连外部存储、哪些是纯计算。
第三步，设置初始并行度。source并行度跟分区数一致，纯计算算子先设为总slot数的70%左右，外部IO算子先设为较小的固定值比如2，然后再调。keyBy算子初始设成与source相同或略低。
第四步，压测观察。用生产环境的流量做压测，打开Flink Web UI，重点看每个算子的“record throughput”和“backpressure”指标。如果某个算子背压很高，说明它处理不过来，先试着提高它的并行度。注意，同时观察上游算子的背压是否减轻。
第五步，微调状态和窗口相关配置。如果用了窗口，窗口时间比较长，比如1小时，那么并行度太高会导致每个窗口的碎片很小，但状态的总数量很多。此时可以适当降低并行度，减少状态复制。
第六步，观察任务稳定性。连续跑几天，看是否有OOM、频繁checkpoint失败、任务重启等问题。如果有，再针对性调整资源或并行度。
这套流程我建议团队里固化下来，每次上线一个新的实时作业都按这个走，能避免很多线上事故。
八、总结
并行度是Flink资源调度的核心参数，它不是越大越好，也不是越小越省，而是要根据数据源特点、计算特性、外部依赖资源来共同决定。从写代码开始，我们就要意识到并行度的优先级关系，知道哪里设置最有效。在产线上，要时刻关注背压、checkpoint和资源使用率，用这些指标倒推并行度是否合理。
真正做调优的时候，心态放稳，不要指望一蹴而就。先定一个合理的初始值，然后靠监控反推，持续调整。理论再多，也不如亲手跑一个作业看看实际数据。希望这篇文章能帮你少踩一些并行度的坑，把Flink跑得更顺更稳。
评论
围绕“Flink并行度设置的核心逻辑及生产环境优化方案”参与讨论