58 集团大规模 Storm 任务平滑迁移至 Flink 的秘密

Flink-Storm 是 Flink 官方提供的用于 Flink 兼容 Storm 程序 beta 工具，而且在 Release 1.8 以后去掉相关代码。本文主要讲述 58 实时计算平台如何优化 Flink-Storm 以及基于 Flink-Storm 实现真实场景下大规模 Storm 任务平滑迁移 Flink。

背景

58 实时计算平台旨在为集团业务部门提供稳定高效实时计算服务，主要基于 Storm 和 Spark Streaming 构建，但在使用过程当中也面临一些问题，主要包括 Storm 在吞吐量不足以及多集群带来运维问题，Spark Streaming 又没法知足低延迟的要求。Apache Flink 开源以后，其在架构设计、计算性能和稳定性上体现出的优点，使咱们决定采用 Flink 做为新一代实时计算平台的计算引擎。同时基于 Flink 开发了一站式高性能实时计算平台 Wstream，支持 Flink jar，Stream Sql，Flink-Storm 等多样化任务构建方式。

在完善 Flink 平台建设的同时，咱们也启动 Storm 任务迁移 Flink 计划，旨在提高实时计算平台总体效率，减小机器成本和运维成本。

Storm vs Flink

尽管 Flink 做为高性能计算引擎能够很好兼容 Storm，但在业务迁移过程当中，咱们仍然遇到了一些问题：

1.用户对 Flink 的学习成本；

1. 从新基于 Flink 开发耗费工做量；
2. Stream-SQL 虽然能够知足快速开发减小学习成本和开发工做量但没法知足一些复杂场景。

所以咱们决定采用 Flink 官方提供的 Flink-Storm 进行迁移，在保障迁移稳定性同时无需用户修改 Storm 代码逻辑。

Flink-Storm 原理

经过 Storm 原生 TopologyBuilder 构建好 Storm topology。

FlinkTopology.createTopology(builder) 将 StormTopology 转换为 Flink 对应的 Streaming Dataflow。

SpoutWrapper 用于将 spout 转换为 RichParallelSourceFunction，spout 的OutputFields转换成 source 的T ypeInformation。

BoltWrapper 用于将 bolt 转换成对应的 operator，其中 grouping 转换为对 spout 的 DataStream 的对应操做。

构建完 FlinkTopology 以后，就能够经过 StreamExecutionEnvironment 生成 StreamGraph 获取 JobGraph，以后将 JobGraph 提交到 Flink 运行时环境。

实践

Flink-Storm 做为官方提供 Flink 兼容 Storm 程序为咱们实现无缝迁移提供了可行性，可是做为 beta 版本，在实际使用过程当中存在不少没法知足现实场景的状况，主要包括版本，功能 bug，复杂逻辑兼容，没法支持 yarn 等，下面将主要分为平台层面和用户层面讲述咱们的使用和改进。

平台层面

1. 版本

当前线上使用 Apache Flink 1.6 版本，Flink-Storm 模块基于 Storm 1.0 开发，咱们平台运行 Storm 版本为 0.9.5 和 1.2 。

1.1 对于 Storm 1.2 运行任务，Storm 1.0 API 彻底兼容 1.2 版本，所以只需切换 Flink-Storm 模块依赖的 storm-core 到 1.2.

1.2 对于 Storm 0.9.5 任务，因为 Storm 1.0 API 没法兼容 0.9.5，须要修改依赖 storm-core 为 0.9.5，同时修改 Flink-Storm 模块中全部与 Storm 相关的 API，主要是切换 package 路径。

1.3 从新构建 flink-storm 包 mvn clean package -Dmaven.test.skip=true -Dcheckstyle.skip=true

2.功能

2.1 传递语义保证

Storm 使用 ACK 机制来实现传递语义保证，咱们没有将 Storm 的 ACK 机制移植到Flink-Storm。所以，某些依赖 ACK 机制的功能会受到限制。好比，Kafka spout 将消费状态存储在 ZK，状态的更新须要依赖 ACK 机制，tuple 树结束后，spout 才会触发状态更新，表示这条消息已经被彻底处理，从而实现 at least once 的传递保证。Storm 也提供了at most once 的支持，spout 发送消息后，无需等待 tuple 树结束直接触发状态更新。咱们使用了 Storm 的实现 at most once 的方式，在 Kafka spout 实现 at most once 的基础上，经过实现 Flink Checkpoint 的状态机制，实现了 Flink-storm 任务的 at least once。Storm 任务迁移到 Flink，传递保证不变。

2.2 tick tuple 机制

Storm 使用 tick tuple 机制实现定时功能，消息超时重发、Bolt 定时触发等功能都要依赖 tick tuple 机制。Storm 0.9.5 版本没有实现窗口功能，用户可使用 tick tuple 机制简单实现窗口功能。咱们一样为 Flink-Storm 增长了 tick tuple 机制的支持，使用方式也和 Storm 中使用方式同样，配置 topology.tick.tuple. freq.secs 参数，即开启了 tick tuple 功能。

2.3 多输入下 AllGrouping 支持

AllGrouping 分组方式对应于 Flink 是 Broadcast。如图，bolt-1 有两个输入，这种状况下，原 flink-storm 的实现，spout-2 到 bolt-1 的数据分区的表现形式和Rebalance（Flink 术语）同样，而不是 Broadcast。咱们优化了这种场景，使其数据分组表现和 Storm 中是同样的。

3.Runtime

Flink-Storm 默认支持 local 和 standalong 模式任务提交，没法将任务提交到 yarn 集群，咱们在建设 Flink 集群一开始就选择了 yarn 模式，便于集群资源管理和统一实时计算平台，所以须要自行实现支持 yarn 的 runtime 功能，这里主要涉及 yarn client 端设计。

YARN Client 实现机制

整个模块主要分为四个部分，其中 client 用于调用 Flink-Storm 程序转化接口，获得 Flink jobGraph。配置参数用于初始化 Flink 及 yarn 相关配置，构建运行时环境，命令行工具主要用于更加灵活的管理。yarnClient 主要实现 ApplicationClientProtocol 接口，完成与 ResourceManager 与 ApplicationMaster 的交互，实现 Flink job 提交和监控。

4.任务部署

为便于任务提交和集成到 Wstream 平台，提供相似 Flink 命令行提交方式：

用户层面

1.maven 依赖

平台将编译好的包上传到公司 maven 私服供用户下载对应版本 Flink-Storm 依赖包：

2.代码改动

用户须要将 Storm 提交任务的方式改为 Flink 提交，其余无需变更。

总结

经过对 Fink-Storm 的优化和使用，咱们已经顺利完成多个 Storm 集群任务迁移和下线，在保障实时性及吞吐量的基础上能够节约计算资源 40% 以上，同时借助 yarn 统一管理实时计算平台无需维护多套 Storm 集群，总体提高了平台资源利用率，减轻平台运维工做量。

做者介绍：万石康，来自 58 集团 TEG，后端高级工程师，专一于大数据实时计算架构设计。

---

阅读原文

本文为云栖社区原创内容，未经容许不得转载。