Apache Flink在阿里的使用（译）

Flink是未来大数据实时数据处理领域的首选框架，本文原文是阿里巴巴的搜索团队总监Xiaowei Jiang在Flink Forward 2016大会上分享的内容，后来被记录并移到Flink公司官网的Blog上（注意这个不是社区的官网，原名叫data Artisans，被阿里收购后改为ververica）。文章内容很不错，我这里也是抱着学习和了解的心态翻译了一下这篇文章，英语好的同学建议直接去原文查看，翻译水平有限，如有不足之处，欢迎轻拍。


原文链接：

https://www.ververica.com/blog/blink-flink-alibaba-search
youtube的视频链接：

https://www.youtube.com/watch?v=w9f-440oejg&feature=youtu.be

阿里巴巴是世界上最大的电子商务零售商。在2015年的年度销售总额是3940亿美元，已经超过了eBay加上Amazon两家公司的总额。这其中主要收入来源于搜索和推荐平台。如何才能构建一个强大的电商搜索引擎？ 答案是可以实时的，尽可能的相关和准确的为每个用户提供不同的内容，也就是所谓的千人千面。

在阿里这种公司的量级下，想要做到这一点并不简单，因为很难寻找到一种技术可以处理适配阿里的各种使用场景，而Apache Flink正是这样一种技术。阿里巴巴内部使用的Blink是社区Flink的一个分支，阿里新定制的特性会首先应用在Blink上，并在时机成熟时反馈给社区分支，Flink为阿里搜索基础设施给终端用户带来的尽可能相关和准确的体验提供了重要的技术支持。

在这篇文章中讲述了Flink在阿里搜索中的角色以及为什么选择Flink用在搜索基础架构中。

Flink在阿里搜索
文档创建
搜索的重要部分就是其数据源，在阿里索引的数据由数以百万计的产品列表和相关的产品数据组成。文档创建面临的挑战是因为数据存储在许多不同的地方，所以需要由搜索团队聚集整个相关信息然后去创建一条完整的搜索数据，总的来说，分为3步：

（1）从不同的数据源（mysql，hdfs等）同步所有的产品数据进入hbase集群

（2）在业务逻辑中完成从不同的表中关联数据，并构建出最终的，可被搜索的文档存储到hbase表中

（3）导出整个hbase表的数据或者部分更新的数据到一个文件中，最终推入搜索引擎里面提供搜索。



上面的3步在阿里是由两个不同的pipelines完成，分别是全量索引构建pipeline和增量索引构建pipeline，这也是典型的lambda架构。

在全量的pipeline中，需要处理所有的数据源构建索引，这是一个典型的批处理任务。

在增量的pipeline中，需要处理在全量pipeline任务结束之后发生更新的数据，例如卖家可以修改商品的价格或者描述，或者库存数量可能发生变化等，这种信息必须尽可能快的反映到搜索结果中，增量的pipeline是一个典型的流任务。

搜索算法的实时A/B测试
搜索算法的效果会被定期的测评，用来指导搜索算法的调优或者改进，这种反馈越快越好，所以阿里使用Blink构建了A/B testing框架。通过搜索实时产生的日志，点击或者交易的数据会被实时的收集，然后经过业务逻辑关联，解析和过滤，最终再将数据聚合，并把聚合结果写入Druid（一种数据分析的OLAP引擎），在Druid中就可以写一个非常复杂OLAP分析语句从而在结果中能够看到不同算法的表现。



在线机器学习
在使用搜索随着时间的变换过程中，商品的CTR，剩余库存，点击率会对应发生变换，通过机器学习和推荐可以帮助提供更相关的搜索rank反馈给用户，在这里面可以通过Flink实时的pipeline来完成特征更新从而提高搜索的转化率。

其次，在特殊的节日中，比如双11里面搜索流量暴增，因此之前的训练的模型可能面对突发流量就完全失效了，这时候通过Flink可以提供高效在线机器学习，通过实时数据及时构建数据模型，从而在这种不常见但很重要的的场景里也能提高搜索转化率。



为什么是Flink
在阿里搜索选择Flink作为关键的技术支持时，主要基于下面的4个方面考虑：

（1）灵活性。 通过更高级别的API可以统一搜索的业务逻辑并维护一套代码库包括搜索2个构建pipeline。

（2）一致性。卖家商品数据的变化必须被反映到最终的搜索结果中，所以搜索基础架构团队要求至少一次的处理语义（部分Flink用户case可以提供准确一次的语义）

（3）低延迟。当库存发生变换时，必须尽可能快的可以被搜索。比如，我们不想要一个拥有高搜索权重的商品是一个已经卖完的商品。

（4）性能。阿里巴巴需要处理大量的数据，在这种规模下，处理效率的提升可以显著的节约成本。我们需要一个能够高效处理高吞吐量的框架。

宽泛的说，关于统一批处理和流处理有两种实践模式，第一种方法是以批处理作为出发点，在批处理的之上模拟流，比如典型的Spark Streaming就是这样以模拟微批的方式实现的流处理，但这种方式并不适合需要低延迟的应用，因为这里有一些固定的开销不可避免，比如每个批处有1000个task，那么每次都需要重新建立1000次链接和重新加载状态，在一些场景下，微批的方式因为太耗时而并不能被采用。

Flink则是采用另一种方式，以流处理作为基本出发点，在流处理的基础之上模拟构建批处理，这种情况下微批次其实是特殊情况下一种流的表现形式，并拥有更低的延迟等其他的一些优点。

Blink是什么
Blink是阿里定制维护的一个从社区Flink拉的分支代码，在内部已经稳定在多个集群中，每个集群约有1000台机器。如此大规模的集群，性能是至关重要的，Blink的性能改进主要覆盖两个方面：

1，完善了Table API，可以使同样的SQL代码运行在批处理和流处理中。

2，更加健壮的Blink On Yarn机制，并兼容Flink的API和其生态系统。

Table API的改进
首先增加了对用户自定义函数的支持，可以方便的在Flink代码中加入业务逻辑。同时也增加了stream-to-stream 的join功能以及一些聚合函数的支持。最有兴趣的可能是我们加入支持了流的window的distinct_count功能。



具体可参考：https://cwiki.apache.org/confluence/display/FLINK/FLIP-11%3A+Table+API+Stream+Aggregations

Blink on Yarn
原生的Flink支持两种模式，分别是standalone和Flink on YARN，在Flink on YARN模式下，一个job不能够动态的申请和释放资源，它必须在运行之前获取可用的资源，并且不同的job可能会共享一个JVM进程，这种模式下可能倾向于提高资源利用率而不是资源隔离。在Blink里面不同的job不能够同一个JVM进程里面，从而在job运行和Task执行做到最好的隔离。阿里团队也正在将这一个改进反馈给社区，并且同时也将这一改进集成到其他的资源调度框架里面，比如：Standalone, Yarn, Mesos, Kubernetes等。



具体细节可参考：
https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=65147077

Operator动态调整
在生产环境，有可能随时需要改变operators的并行度，但是同时还不能丢失状态，早期的Flink并不支持在维护状态的同时改变operators的并行度，Blink通过引入buckets的概念，来使得这个功能得以实现。这儿有许多的buckets和tasks，每个task可以被分配到多个buckets里面，当并行度改变的时候会重新分配task到buckets里面，并且不影响状态的维护。Flink在1.2的版本引入了key groups的概念来解决了这个问题，理论上与buckets方式相同，具体细节可参考：
https://issues.apache.org/jira/browse/FLINK-3755



增量Checkpoint
在Flink里面，checkpoint发生在2个阶段：先在本地生成一个快照状态，然后持久快照的状态到HDFS或者其他的外部存储，每个快照都是完整的。在阿里这个快照状态的体积异常庞大，所以不能使用这种方式。故而在Blink里面仅仅存储有修改的状态到HDFS里面并对性能做了优化，这个修改最终使得阿里可以在生产环境使用大的checkpoint状态。



异步IO
在阿里来自生产环境的一个瓶颈是许多Flink任务要频繁的访问外部的存储系统比如Hbase，为了解决这个问题，引入了异步IO操作。

具体细节可参考：
https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=65870673



Flink在阿里的下一步计划
阿里会持续的优化相关的流处理任务，尤其是在任务的临时倾斜，慢机器的背压策略以及如何快速的从失败中恢复。 Flink基于流的批处理机制，在未来会有巨大的潜力，阿里在这方面会重点投入，期望在未来可以让Flink在生产环境有一个批处理模型。

另外一个流行话题是关于streaming SQL，阿里也会持续的在Flink中添加到SQL和Table API支持。当然阿里集群的规模数据也在不断的增加，如何解决大规模集群下的扩展会是会变得更加重要，阿里也在不断的与社区沟通和协同，并将Blink新的特性反馈给社区以便于所有的Flink用户可以得到这种便利。