Flink教程（28）- Flink性能优化

01 引言

在前面的博客，我们学习了Flink的Metrics监控了，有兴趣的同学可以参阅下：

• 《Flink教程（01）- Flink知识图谱》
• 《Flink教程（02）- Flink入门》
• 《Flink教程（03）- Flink环境搭建》
• 《Flink教程（04）- Flink入门案例》
• 《Flink教程（05）- Flink原理简单分析》
• 《Flink教程（06）- Flink批流一体API（Source示例）》
• 《Flink教程（07）- Flink批流一体API（Transformation示例）》
• 《Flink教程（08）- Flink批流一体API（Sink示例）》
• 《Flink教程（09）- Flink批流一体API（Connectors示例）》
• 《Flink教程（10）- Flink批流一体API（其它）》
• 《Flink教程（11）- Flink高级API（Window）》
• 《Flink教程（12）- Flink高级API（Time与Watermaker）》
• 《Flink教程（13）- Flink高级API（状态管理）》
• 《Flink教程（14）- Flink高级API（容错机制）》
• 《Flink教程（15）- Flink高级API（并行度）》
• 《Flink教程（16）- Flink Table与SQL》
• 《Flink教程（17）- Flink Table与SQL（案例与SQL算子）》
• 《Flink教程（18）- Flink阶段总结》
• 《Flink教程（19）- Flink高级特性（BroadcastState）》
• 《Flink教程（20）- Flink高级特性（双流Join）》
• 《Flink教程（21）- Flink高级特性（End-to-End Exactly-Once）》
• 《Flink教程（22）- Flink高级特性（异步IO）》
• 《Flink教程（23）- Flink高级特性（Streaming File Sink）》
• 《Flink教程（24）- Flink高级特性（File Sink）》
• 《Flink教程（25）- Flink高级特性（FlinkSQL整合Hive）》
• 《Flink教程（26）- Flink多语言开发》
• 《Flink教程（27）- Flink Metrics监控》

本文主要讲解Flink的性能优化。

02 History Server

flink的HistoryServer主要是用来存储和查看任务的历史记录，具体信息可以看官网： https://ci.apache.org/projects/flink/flink-docs-release-1.12/deployment/advanced/historyserver.html

相关配置：

# Directory to upload completed jobs to. Add this directory to the list of
# monitored directories of the HistoryServer as well (see below).
# 将已完成的作业上传到的目录
jobmanager.archive.fs.dir: hdfs://node01:8020/completed-jobs/

# The address under which the web-based HistoryServer listens.
# 基于 Web 的 HistoryServer 的地址
historyserver.web.address: 0.0.0.0

# The port under which the web-based HistoryServer listens.
# 基于 Web 的 HistoryServer 的端口号
historyserver.web.port: 8082

# Comma separated list of directories to monitor for completed jobs.
# 以逗号分隔的目录列表，用于监视已完成的作业
historyserver.archive.fs.dir: hdfs://node01:8020/completed-jobs/

# Interval in milliseconds for refreshing the monitored directories.
# 刷新受监控目录的时间间隔（以毫秒为单位）
historyserver.archive.fs.refresh-interval: 10000

参数解析：

• jobmanager.archive.fs.dir：flink job运行完成后的日志存放目录
• historyserver.archive.fs.dir：flink history进程的hdfs监控目录
• historyserver.web.address：flink history进程所在的主机
• historyserver.web.port：flink history进程的占用端口
• historyserver.archive.fs.refresh-interval：刷新受监视目录的时间间隔（以毫秒为单位）。

默认启动端口8082：

• bin/historyserver.sh (start|start-foreground|stop)

03 序列化

Java 原生的序列化方式：

• 优点：好处是比较简单通用，只要对象实现了 Serializable接口即可；
• 缺点：效率比较低，而且如果用户没有指定 serialVersionUID的话，很容易出现作业重新编译后，之前的数据无法反序列化出来的情况（这也是 Spark Streaming Checkpoint的一个痛点，在业务使用中经常出现修改了代码之后，无法从 Checkpoint 恢复的问题）

对于分布式计算来讲，数据的传输效率非常重要。好的序列化框架可以通过较低的序列化时间和较低的内存占用大大提高计算效率和作业稳定性。

在数据序列化上，Flink 和 Spark 采用了不同的方式：

• Spark 对于所有数据默认采用 Java 原生序列化方式，用户也可以配置使用 Kryo；相比于 Java 原生序列化方式，无论是在序列化效率还是序列化结果的内存占用上，Kryo 则更好一些（Spark 声称一般 Kryo 会比 Java 原生节省 10x 内存占用）；Spark 文档中表示它们之所以没有把 Kryo 设置为默认序列化框架的唯一原因是因为 Kryo 需要用户自己注册需要序列化的类，并且建议用户通过配置开启 Kryo。
• Flink 则是自己实现了一套高效率的序列化方法。

04 复用对象

比如如下代码：

stream
    .apply(new WindowFunction() {
        @Override
        public void apply(String userName, TimeWindow timeWindow, Iterable iterable, Collector collector) throws Exception {
            long changesCount = ...
            // A new Tuple instance is created on every execution
            collector.collect(new Tuple2(userName, changesCount));
        }
    }

可以看出，apply函数每执行一次，都会新建一个Tuple2类的实例，因此增加了对垃圾收集器的压力。解决这个问题的一种方法是反复使用相同的实例：

stream
        .apply(new WindowFunction() {
    // Create an instance that we will reuse on every call
    private Tuple2 result = new Tuple();
    @Override
    public void apply(String userName, TimeWindow timeWindow, Iterable iterable, Collector collector) throws Exception {
        long changesCount = ...
        // Set fields on an existing object instead of creating a new one
        result.f0 = userName;
        // Auto-boxing!! A new Long value may be created
        result.f1 = changesCount;
        // Reuse the same Tuple2 object
        collector.collect(result);
    }
}

这种做法其实还间接创建了Long类的实例。

为了解决这个问题，Flink有许多所谓的value class:IntValue、LongValue、StringValue、FloatValue等。下面介绍一下如何使用它们：

stream
        .apply(new WindowFunction() {
    // Create a mutable count instance
    private LongValue count = new LongValue();
    // Assign mutable count to the tuple
    private Tuple2 result = new Tuple("", count);

    @Override
    // Notice that now we have a different return type
    public void apply(String userName, TimeWindow timeWindow, Iterable iterable, Collector collector) throws Exception {
        long changesCount = ...

        // Set fields on an existing object instead of creating a new one
        result.f0 = userName;
        // Update mutable count value
        count.setValue(changesCount);

        // Reuse the same tuple and the same LongValue instance
        collector.collect(result);
    }
}

05 数据倾斜

我们的flink程序中如果使用了keyBy等分组的操作，很容易就出现数据倾斜的情况，数据倾斜会导致整体计算速度变慢，有些子节点甚至接受不到数据，导致分配的资源根本没有利用上。

带有窗口的操作:

• 带有窗口的每个窗口中所有数据的分布不平均，某个窗口处理数据量太大导致速率慢
• 导致Source数据处理过程越来越慢
• 再导致所有窗口处理越来越慢

不带有窗口的操作:

• 有些子节点接受处理的数据很少，甚至得不到数据，导致分配的资源根本没有利用上

WebUI体现： WebUI中Subtasks中打开每个窗口可以看到每个窗口进程的运行情况：如上图，数据分布很不均匀，导致部分窗口数据处理缓慢。

优化方式：

• 对key进行均匀的打散处理（hash，加盐等）
• 自定义分区器
• 使用Rebalabce

注意：Rebalance是在数据倾斜的情况下使用，不倾斜不要使用，否则会因为shuffle产生大量的网络开销。

06 总结

本文主要从History Server、序列化、 复用对象、 数据倾斜来讲解了Flink的性能优化，谢谢大家的阅读，本文完！