Flink-Storm在58的应用实践

简介

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的学习成本；

2. 重新基于Flink开发耗费工作量；

3. Stream-Sql虽然可以满足快速开发减少学习成本和开发工作量但无法满足一些复杂场景。

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

Flink-Storm原理

1.通过Storm原生TopologyBuilder构建好storm topology。

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

3.SpoutWrapper用于将spout转换为RichParallelSourceFunction，spout的OutputFields转换成source的TypeInformation。

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

5.构建完FlinkTopology之后，就可以通过StreamExecutionEnvironment生成StreamGraph获取JobGraph，之后将JobGraph提交到Flink运行时环境。

实践

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

平台层面

1. 版本

当前线上使用apache flink1.6版本，Flink-Storm模块基于Storm 1.0开发，我们平台运行Storm版本为0.9.5和1.2 。

1.1 对于Storm 1.2运行任务，Storm1.0 api完全兼容1.2版本，因此只需切换Flink-Storm模块依赖的storm-core到1.2

1.2 对于Storm0.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架构师