腾讯实时计算团队为业务部门提供高效、稳定和易用的实时数据服务。其每秒接入的数据峰值达到了 2.1 亿条，每天接入的数据量达到了 17 万亿条，每天的数据增长量达到了 3PB，每天需要进行的实时计算量达到了 20 万亿次。腾讯选择用 Flink 作为新一代的实时流计算引擎，并对社区版的 Flink 进行了深度的优化，在此之上构建了一个集开发、测试、部署和运维于一体的一站式可视化实时计算平台——Oceanus。本文整理自腾讯大数据高级工程师杨华在 QCon全球软件开发大会（北京站）2019上的演讲，他介绍了腾讯流计算技术的演进过程，产品化及云端对外服务进展。

大家好，我是来自腾讯大数据团队的杨华（vinoyang），很高兴能够参加这次北京的QCon，有机会跟大家分享一下腾讯实时流计算平台的演进与这个过程中我们的一些实践经验。

这次分享主要包含四个议题，我会首先阐述一下腾讯在实时计算中使用Flink的历程，然后会简单介绍一下腾讯围绕Flink的产品化实践：我们打造了一个Oceanus平台，同时腾讯云也早已提供基于Flink的实时流计算服务，接着我们会重点跟大家聊一聊我们对社区版Flink的一些扩展与改进、优化。

Flink在腾讯实时计算概况简介

首先，我们进入第一个议题。Flink在腾讯正式被考虑替代Storm是在2017年。

17年上半年，我们主要在调研Flink替换Storm的可行性、特性、性能是否能够满足我们的上线要求。在此之前，我们内部以Storm作为实时计算的基础框架也已经有几年的时间了，在使用的过程中也发现了Storm的一些痛点，比如，没有内置状态的支持，没有提供完备的容错能力，没有内置的窗口API，core API无法提供Exactly-once的语义保证等等。

17年下半年，我们从社区拉出当时最新的发布分支（1.3.2）作为我们内部的定制开发分支进行开发。作为一个试点，我们选择了内部一个流量较大的业务来进行替换，这个业务在我们内部是以standalone的模式部署的，所以我们最初也使用的是Flink的standalone部署模式。

18年上半年，我们开始围绕Flink进行产品化，打造了一个全流程、一体化的实时流计算平台——Oceanus，来简化业务方构建实时应用的复杂度并降低运维成本，这也基本明确了后续我们主要的运行模式是Flink on YARN。

18年下半年，我们的Oceanus平台已经有足够的能力来构建常见的流计算应用，我们部门内部的一些实时流计算业务也已经在平台上稳定运行，于是我们开始为腾讯云、腾讯其他事业群以及业务线提供流计算服务。同时，我们也将平台整合进我们的大数据套件，为外部私有云客户提供流计算服务。

19年上半年，我们的主要目标是在Oceanus上沉淀并完善上层的场景化服务建设，比如提供在线机器学习、风控等场景化服务。另外，我们也在 Flink 批处理方向发力，利用Flink的计算能力来满足跨数据中心，跨数据源的联合分析需求。它可以做到：数据源SQL下推，避免集群带宽资源浪费；单DC内CBO（基于代价优化），生成最优的执行计划；跨DC CBO，根据DC负载和资源选择最佳DC执行计算，从而获得更好的资源利用和更快的查询性能。以上就是腾讯使用Flink的整个历程。

这幅图展示了，Flink目前在腾讯内部已经为一些我们耳熟能详的产品提供实时计算的服务。这些产品，包括微信、支付、财付通、腾讯云、QQ、空间、音乐、游戏、K歌等等。我们列举其中几个业务的使用案例，微信使用我们的平台来统计朋友圈的实时浏览信息、小游戏种子用户的UV计算、实时恶意流量分析判断、看一看的红点信息；支付用来计算商户交易相关的统计；音乐用于实时点唱、热门排行榜等等。

接下来我们来了解一下，目前Flink在腾讯使用的现状。目前我们Oceanus平台YARN集群的vcore总数目达到了34万，累计的峰值计算能力接近2.1亿/秒，日均处理消息规模近20万亿。到目前为止，腾讯内部除了广告的在线训练业务外，原先运行在Storm上的实时流计算业务都已逐步迁移到Flink引擎上，而广告这块的业务预计也会在今年下半年迁移完成。

Oceanus平台简介

接下来，我们进入第二个议题：简要介绍一下我们的Oceanus平台。

首先，我们来看一下Oceanus平台的整体技术架构。我们内部定制版的Flink引擎，称之为TDFLINK，它跟其他的一些大数据基础设施框架交互并协同支撑了我们上层的Oceanus平台，Oceanus支持画布、SQL以及Jar三种形式构建应用，为了方便业务方降低整体成本，我们还提供了配置、测试、部署等完整配套的功能，在平台之上我们提供了一些领域特定的场景化服务比如ETL、监控、推荐广告等。

下面我们来介绍Oceanus的几个典型功能。首先这是某个用户的应用列表页。从列表中，我们可以看到应用的当前状态、类型、迭代的版本，它归属于哪个场景等信息。在操作栏，我们可以一键对应用进行启停、调试，查看它的指标信息等，除此之外我们还提供了很多便捷的操作，比如快速复制一个应用，他们都收纳在“更多”菜单按钮中。

这是我们的一个指标分钟级统计的画布应用详情页，我们为ETL类型的应用提供了一个通用的transform算子。它提供了很多功能细分的可插拔的便捷函数来简化常见的事件解析与提取的复杂度。在图中，多种不同类型的指标经过split算子分流后将相同的指标进行归类，然后再对它们应用各自的统计逻辑，就像这里的窗口一样，基本上每个算子都是配置化的。像这种类型的应用我们通过拖拽、配置就可以轻松完成它的构建。

这幅图展示了我们的指标详情页检查点的指标明细，为了让业务人员更直观地了解它们最关心的指标信息，我们将一些必要的指标进行了重新梳理并展示到我们的平台上，这里有些指标直接使用了Flink提供的REST API接口，而有些指标则是我们内部扩展定制的。

最后，我们来介绍一下最近上线的在线机器学习模块。这是我们一个模型训练应用的详情页，同样它也是画布类型的，我们对常规的机器学习类型的应用进行了步骤拆分，包括了数据预处理、算法等相关步骤，每个步骤都可以进行拖拽，再加上配置的方式就可以完成一个ML类型的应用创建。

对于训练得到的模型，我们也提供了模型服务功能，我们用模型服务组来管理每个模型的不同时间的版本，点击右侧的“评估报告”可以查看这个模型的AUC趋势。

以上是对Oceanus平台的介绍，如果大家有兴趣可以扫描PPT最后的二维码来进一步了解我们的平台以及腾讯云上的流计算服务。

针对Flink的扩展与优化

接下来，我们进入下一个议题，介绍我们内部Flink版本在通过腾讯云对外提供服务时基于内部以及业务的相关需求对社区版的扩展与优化。

第一个改进是我们重构了Flink Web UI，我们重构的原因是因为社区版的Flink Web UI在定位问题的时候不能提供足够的信息，导致问题定位的效率不够高。尤其是Job并行度非常大，YARN的container数目非常多的时候，当Job发生失败，很难快速去找到container和节点以查看进程的堆栈或者机器指标。所以，为了更高效地定位问题，我们对Flink web UI进行了重构并暴露了一些关键指标。

这是我们一个TaskManager的详情页，我们为它新增了一个“Threads” Tab，我们可以通过它看到Task相关的线程信息：线程名称、CPU消耗、状态以及堆栈等。这样一旦哪个算子的线程可能成为瓶颈时，我们可以快速定位到它阻塞在什么方法调用上。

接下来的这个改进是对JobManager failover的优化。大家应该都知道社区版的Flink JobManager HA在Standalone模式下有个很大的问题是：它的standby节点是冷备的，JobManager的切换会导致它管理的所有Job都会被重启恢复，这一行为在我们现网环境中是不可接受的。所以，我们首先定制的第一个大特性就是JobManager的failover优化，让standby节点变成热备，这使得JobManager的切换对TaskManager上已经正在运行的Job不产生影响。我们已经对Standalone以及Flink on YARN这两种部署模式支持了这个特性，Flink on YARN的支持还处于内部验证阶段。我们以对Standalone模式的优化为例来进行分析，它主要包含这么几个步骤：

取消JobManager跟TaskManager因为心跳超时或Leadership变动就cancel task的行为；
对ExecutionGraph核心数据的快照；
通过ExecutionGraphBuilder重构空的ExecutionGraph加上快照重置来恢复出一个跟原先等价的ExecutionGraph对象；
TaskManager跟新的JobManager leader建立连接后以心跳上报自己的状态和必要的信息；
新的JobManager确认在reconcile阶段Job的所有task是否正常运行；

接下来的这个改进已经在反馈社区的过程中，它就是对检查点失败处理的改进。在探讨改进之前，我们先来了解一下社区版当前的处理机制。JobMaster中，每个Job会对应一个Checkpoint Coordinator，它用来管理并协调Job检查点的执行。当到达一个检查点的触发周期，Coordinator会对所有的Source Task下发TriggerCheckpoint消息，source task会在自身完成快照后向下游广播CheckpointBarrier，作为下游task触发的通知。其中，如果一个task在执行检查点时失败了，这取决于用户是否容忍这个失败（通过一个配置项），如果选择不容忍那么这个失败将变成一个异常导致task的失败，与此同时task的失败将会通知到JobMaster，JobMaster将会通知这个Job的其他task取消它们的执行。现有的机制存在一些问题：

Coordinator并不能控制Job是否容忍检查点失败，因为控制权在task端；
Coordinator当前的失败处理代码逻辑混乱，区分出了触发阶段，却忽略了执行阶段；
无法实现容忍多少个连续的检查点失败则让Job失败的逻辑。

了解了现有的实现机制，我们再来看接下来的改进方案。首先，我们对源码中checkpoint package下的相关类进行了重构，使得它不再区分触发阶段，引进了更多的检查点失败原因的枚举并重构了相关的代码。然后我们引入了CheckpointFailureManager组件，用来统一失败管理，同时为了适配更灵活的容忍失败的能力，我们引入了检查点失败计数器机制。现在，当我们遇到检查点失败后，这个失败信息会直接上报到Coordinator，而是否要让Job失败具体的决策则由CheckpointFailureManager作出，这就使得Coordinator具有了完整的检查点控制权，而决策权转让给CheckpointFailureManager，则充分实现了逻辑解耦。

下面我们要看的这个特性是对Flink原生窗口的增强，所以我们称之为Enhanced window。大家都知道Flink的EventTime语义的窗口无法保证任意延迟到达的数据都能参与窗口计算，它只允许你设置一个容忍延迟的时间。但我们的应用场景里，数据的延迟可能非常高，甚至有时跨天的也会发生，但我们无法为常规的窗口设置这么长的延迟时间，并且我们的业务无法容忍延迟数据被丢弃的行为。因此针对这种场景，Flink自带的窗口无法满足我们的需求。所以，我们对它做了一些改进，它能够容忍任意延迟到来的事件，所有的事件都不会被丢弃，而是会加入一个新的窗口重新计算，新窗口跟老窗口毫无关系，所以最终可能针对一个窗口在用户的目标表中会存在多条记录，用户只需自行聚合即可。

为了方便在上层使用这种窗口，我们为它定制了SQL关键字，这幅图展示了我们在指标统计场景中使用它的一个示例。

这是我们根据业务需求所定制的另一个窗口——增量窗口。在业务中经常遇到这样的需求：希望看到一个窗口周期内的增量变化，这个窗口周期可能会很长，比如一个天级别的窗口。比如我们希望看到一天内每个小时的PV增长趋势，或者游戏中的一些虚拟物品的消耗趋势。Flink默认的翻滚窗口以及触发器是没有内置这种窗口内小批次触发的功能。当然我们也可以通过一个个的小窗口来计算阶段性的结果，然后再对数据进行二次处理，但这样会比较麻烦。所以我们实现了大窗口内多次增量触发的功能，扩展实现了一个窗口内多次触发的Trigger，并定制了相应的SQL语法来供业务使用。这里我们可以看到虽然是大窗口，但由于数据都在不断地进行增量聚合，所以并不会hold住非常大的状态集。

这幅图展示了增量窗口的使用方式，通过新的关键字，底层会映射到我们自实现的触发器。

接下来我们要看的这个特性是我们对Flink keyBy的优化，我们称之为：LocalKeyBy。我们在使用KeyBy的时候都遇到过数据热点的问题，也就是数据倾斜。数据倾斜主要是业务数据的key取值不够离散，而keyBy背后是hash的partition方式，它根据一个key的hash值来决定数据要落到哪个节点分区。如果发生数据倾斜很容易造成计算资源利用不均以及反压（back pressure）等问题产生。针对这一点，我们在保证计算语义的情况下对keyBy进行了优化，开发了LocalKeyBy功能。它的原理是通过本地预聚合来减少发送的数据量，但这里需要注意的一点是：使用这个算子的时候，需要对原有的实现代码进行调整，因为它将原来的keyBy拆分为了两步：预聚合以及合并。

我们在本地对keyBy与LocalKeyBy做了一个简单的性能对比测试，发现在流量倾斜严重的情况下，使用LocalKeyBy整体性能并没有受到太大的影响，但Flink原生的keyBy则随着流量的倾斜而产生显著的性能下降。

我们继续来看一个特性：水位线算子定时检测流分区空闲的功能。Flink社区目前针对Source实现了定时的流idle检测功能（虽然没有开放），它主要针对的场景是Kafka某个分区空闲无数据从而造成对应的subtask无法正常提取watermark，导致对下游的计算产生影响。

但我们的场景和社区略有差别，我们没有将所有的逻辑都压到source里，为了进行逻辑拆分我们引入了一个transform算子，它专门针对ETL的场景，所以我们的watermark很多情况下不在source算子上提取，而是在下游的某个算子上，在某些情况下，如果watermark的分配算子在filter之类的算子后面，则可能造成某个pipeline在中间断流，也造成了无法正常提取watermark的情形。针对这种场景，我们在提取watermark的算子上也实现了定时检测流idle的功能。这样就算因为某个分区的数据都被过滤掉造成空闲，也不至于对下游的计算产生影响。

我们介绍的下一个特性是Framework与用户业务日志的分离。这个特性其实最受益的是Standalone部署模式，因为这种模式下多job的task是混合部署在同一个TaskManager中的，而TaskManager本身只使用一个日志文件来记录日志。所以，这导致排查业务问题非常麻烦。另外，我们对Flink web UI展示日志文件也做了一些改进，我们会列出JobManager以及TaskManager的日志文件夹中所有的文件列表。这是因为，随着流应用长时间运行，累积的日志量会越来越大，我们通常都会对应用的日志配置滚动策略，除此之外我们还会输出GC日志等，而Flink的web UI默认只能展示最新的那个日志文件，这对于我们定位问题很不方便。所以，我们引入了一个新的tab，它能够列出日志文件夹下的文件列表，点击后再请求特定的日志文件。

在分析这个特性的实现之前，我们需要先了解Flink目前加载日志框架类的方式，它为了避免跟业务Job中可能包含的日志框架的依赖、配置文件产生冲突，日志相关类的加载都代理给平台的类加载器，也就是TaskManager的类加载器，而TaskManager本身加载的这些类都是从Flink安装包的lib底下加载的。而关于日志配置文件，Flink通过JVM启动参数来指定配置文件路径以及日志文件路径。这些机制共同保证了Flink不会受到用户job jar的干扰。

所以，如果我们要实现日志分离，我们就需要打破Flink原先的实现机制，关键点在于：为不同Job的Task加载不同的日志类；为不同Job的Task指定不同的配置文件以及用户日志文件的路径。这意味着我们需要定制Flink自带的user classloader。针对第一点，我们不再将这些日志类的加载代理给平台的加载器，而是将平台类加载器中日志相关的jar的classpath加入到各个task自己的classloader中。关于配置文件，我们显然也不能用Flink平台的配置文件。我们会拿平台使用的配置文件作为模板，对其内部的日志路径进行动态修改，然后将内存中的这个配置文件传递给特定的日志框架。那么这里就有一个问题，内存中的配置文件二进制数据怎么被日志框架读取。log4j以及logback都可以接收配置文件的URL表示，而URL也可以接收一个URLStreamHandler的实现（它是所有流协议的处理器用于连接二进制的数据流与URL），通过效仿bytebuddy（一个动态修改Java二进制字节码的类库），我们实现了ByteArrayUrlStreamHandler来进行二进制的配置文件跟URL之间的衔接，这两点完成后不同Job的Task的类加载器就保证了日志类加载和配置的完全独立性。

目前，我们内部所定制优化的一些特性有些已逐步反馈给社区，还有一些比较大的改动也在跟社区商讨合并计划。我们欢迎有志于迎接万亿级数据规模挑战以及参与Flink 引擎研发的同学加入我们。

总结

以上就是所有的分享内容，由于表达与时间关系，可能很多点不能很好地阐述清楚，如果大家有想法我们可以再线下交流。也欢迎大家扫码了解腾讯的大数据产品、腾讯云上的流计算服务以及Oceanus的功能。

演讲嘉宾

杨华（vinoyang）腾讯大数据高级工程师，参与腾讯实时流计算平台 Oceanus 底层 Flink 引擎的研发与维护，经历了 Flink 在腾讯从零开始落地到目前支撑日均近 20 万亿消息处理规模的全过程。拥有多年的分布式、大数据系统研发经验，Flink 国内较早的布道者之一，也是 Flink 及多个 Apache 项目的长期活跃贡献者。

QCon 全球软件开发大会（北京站）2019 已经圆满结束，QCon上海2019 即将起航，点击此处了解详情。大会7折早鸟票限时开售，现在报名立减2640元，团购可享更多优惠！有任何问题欢迎联系票务小姐姐 Ring ，电话：13269076283 微信：qcon-0410

创作场景

腾讯基于 Flink 的实时流计算平台演进之路