生产总结：Flink应用生产状态有哪些设置

问题导读
1.Flink最大并行度如何设置？
2.产生中Flink状态后端最好使用什么？
3.配置 JobManager 的高可用性有什么好处？

本文阐述了使 Flink 应用达到生产就绪状态所需要的一些配置步骤。在以下部分中，我们概述了重要的配置参数，这些参数是技术领导、DevOps、工程师们在将 Flink 应用程序上线生产之前都需要仔细考虑的。Apache Flink 为大多数配置都提供了开箱即用的默认选项，在许多情况下，它们是POC阶段（概念验证）或探索 Flink 不同 API 和抽象的很好的起点。

然而，将 Flink 应用程序投入生产还需要额外的配置，这些配置可以高效地调整应用的规模，使其达到生产就绪状态，并能与不同系统之间保持兼容，以保证未来迭代升级的需求。

下面几点是我们收集的需要在 Flink 应用上线前做的检查：



1. 明确定义 Flink 算子的最大并发度
Flink 的 keyed state 是由 key group 进行组织的，并分布在 Flink 算子（operator）的各个并发实例上。Key group 是用来分布和影响 Flink 应用程序可扩展性的最小原子单元，每个算子上的 key group 个数即为最大并发数（maxParallelism），可以手动配置也可以直接使用默认配置。默认值粗略地使用 operatorParallelism * 1.5 ，下限 128，上限 32768 。可以通过 setMaxParallelism(int maxParallelism) 来手动地设定作业或具体算子的最大并发。

任何进入生产的作业都应该指定最大并发数。但是，一定要仔细考虑后再决定该值的大小。因为一旦设置了最大并发度（无论是手动设置，还是默认设置），之后就无法再对该值做更新。想要改变一个作业的最大并发度，就只能将作业从全新的状态重新开始执行。目前还无法在更改最大并发度后，从上一个 checkpoint 或 savepoint 恢复。

最大并发度的取值建议设定一个足够高的值以满足应用未来的可扩展性和可用性，同时，又要选一个相对较低的值以避免影响应用程序整体的性能。这是由于一个很高的最大并发度会导致 Flink 需要维护大量的元数据（用于扩缩容），这可能会增加 Flink 应用程序的整体状态大小。

有些老铁对Flink半生不熟，所以上面可能会比较懵，这里举例，如何设置最大并行度。

如下面Java代码

[mw_shl_code=java,true]final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setMaxParallelism(8)
DataStream<String> text = [...]
DataStream<Tuple2<String, Integer>> wordCounts = text
    .flatMap(new LineSplitter())
    .keyBy(0)
    .timeWindow(Time.seconds(5))
    .sum(1).setParallelism(5);

wordCounts.print();

env.execute("Word Count Example");[/mw_shl_code]

上面env.setMaxParallelism(8)为设置最大并行度。setParallelism(5)设置并行度为5，这个是可行的。如果设置为9，则会报错

2. 为 Flink 算子指定唯一用户ID（UUID）
对于有状态的 Flink 应用，推荐给每个算子都指定唯一用户ID（UUID）。 严格地说，仅需要给有状态的算子设置就足够了。但是因为 Flink 的某些内置算子（如 window）是有状态的，而有些是无状态的，可能用户不是很清楚哪些内置算子是有状态的，哪些不是。所以从实践经验上来说，我们建议每个算子都指定上 UUID。

Flink 算子的 UUID 可以通过 uid(String uid) 方法指定。算子 UUID 使得 Flink 有效地将算子的状态从 savepoint 映射到作业修改后（拓扑图可能也有改变）的正确的算子上，这是 savepoint 在 Flink 应用中正常工作的一个基本要素。

如果UUID不设置，那么如何一旦代码改动，就可能状态不能正确的恢复

3. 充分考虑 Flink 程序的状态后端
当前 Flink 还不支持状态后端之间的互换功能，也就是当我们用内存状态后端做了一个 savepoint，我们无法把作业改成 RocksDB 状态后端然后恢复。所以，开发人员和工程负责人在将作业投向生产之前要仔细考虑好该 Flink 应用的最合适的状态后端类型。

关于 Flink 当前支持的三种不同的状态后端类型，对于生产用例来说，强烈建议使用 RocksDB 状态后端，因为这是目前唯一一种支持大型状态和异步操作（如快照过程）的状态后端，异步操作能使 Flink 不阻塞正常数据流的处理的情况下做快照操作。另一方面，使用 RocksDB 状态后端可能存在性能折衷，因为所有状态访问和检索都需要序列化（和反序列化）来跨越 JNI 边界，这与内存状态后端相比可能会影响应用程序的吞吐量。

状态后端，其实就是对状态采用什么样的存储方式，比如使用内存，还是hdfs，还是RocksDB。RocksDB可以近似理解数据库。

4. 配置 JobManager 的高可用性（HA）
高可用性（HA）配置确保了 Flink 应用中 JobManager 组件发生潜在故障后的自动恢复，从而将停机时间降到最低。JobManager 的主要职责是协调 Flink 的部署，例如调度和适当的资源分配。

默认情况下，Flink 为每个集群设置一个 JobManager 实例。这会导致单点故障（SPOF）：如果 JobManager 崩溃了，则无法提交新的作业，而且正在运行的程序也会失败。因此，强烈建议为生产用例配置高可用性（HA）。

上述 4 个步骤总结自社区的最佳实践，使得 Flink 应用能够保持状态的同时任意地扩缩容，处理更大规模的数据和状态，并提高系统的可用性。我们强烈建议您在将应用投入生产之前，仔细阅读上述步骤。

最新经典文章，欢迎关注公众号

原文链接http://wuchong.me/blog/2018/12/03/flink-tips-4-steps-flink-application-production-ready/