京东|京东：Flink SQL 优化实战

本文作者为京东算法服务部的张颖和段学浩，并由 Apache Hive PMC ，阿里巴巴技术专家李锐帮忙校对。主要内容为：
1.背景
2.Flink SQL 的优化
3.总结
一、背景
目前，京东搜索推荐的数据处理流程如上图所示。可以看到实时和离线是分开的，离线数据处理大部分用的是 Hive / Spark ，实时数据处理则大部分用 Flink / Storm 。
这就造成了以下现象：在一个业务引擎里，用户需要维护两套环境、两套代码，许多共性不能复用，数据的质量和一致性很难得到保障。且因为流批底层数据模型不一致，导致需要做大量的拼凑逻辑；甚至为了数据一致性，需要做大量的同比、环比、二次加工等数据对比，效率极差，并且非常容易出错。
而支持批流一体的 Flink SQL 可以很大程度上解决这个痛点，因此我们决定引入 Flink 来解决这种问题。
在大多数作业，特别是 Flink 作业中，执行效率的优化一直是 Flink 任务优化的关键，在京东每天数据增量 PB 级情况下，作业的优化显得尤为重要。
写过一些 SQL 作业的同学肯定都知道，对于 Flink SQL 作业，在一些情况下会造成同一个 UDF 被反复调用的情况，这对一些消耗资源的任务非常不友好；此外，影响执行效率大致可以从 shuffle、join、failover 策略等方面考虑；另外， Flink 任务调试的过程也非常复杂，对于一些线上机器隔离的公司来说尤甚。
为此，我们实现了内嵌式的 Derby 来作为 Hive 的元数据存储数据库 (allowEmbedded)；在任务恢复方面，批式作业没有 checkpoint 机制来实现failover ，但是 Flink 特有的 region 策略可以使批式作业快速恢复；此外，本文还介绍了对象重用等相关优化措施。
二、 Flink SQL 的优化 1. UDF 重用
在 Flink SQL 任务里会出现以下这种情况：如果相同的 UDF 既出现在 LogicalProject 中，又出现在 Where 条件中，那么 UDF 会进行多次调用 (见https://issues.apache.org/jira/browse/FLINK-20887) 。但是如果该 UDF 非常耗 CPU 或者内存，这种多余的计算会非常影响性能，为此我们希望能把 UDF 的结果缓存起来下次直接使用。在设计的时候需要考虑：(非常重要：请一定保证 LogicalProject 和 where 条件的 subtask chain 到一起)
一个 taskmanager 里面可能会有多个 subtask ，所以这个 cache 要么是 thread (THREAD LOCAL) 级别要么是 tm 级别；为了防止出现一些情况导致清理 cache 的逻辑走不到，一定要在 close 方法里将 cache 清掉；为了防止内存无限增大，选取的 cache 最好可以主动控制 size；至于 “超时时间” ，建议可以配置一下，但是最好不要小于 UDF 先后调用的时间；上文有提到过，一个 tm 里面可能会有多个 subtask ，相当于 tm 里面是个多线程的环境。首先我们的 cache 需要是线程安全的，然后可根据业务判断需不需要锁。根据以上考虑，我们用 guava cache 将 UDF 的结果缓存起来，之后调用的时候直接去cache 里面拿数据，最大可能降低任务的消耗。下面是一个简单的使用（同时设置了最大使用 size、超时时间，但是没有写锁）：
public class RandomFunction extends ScalarFunction { private static CacheString Integercache = CacheBuilder.newBuilder() .maximumSize(2) .expireAfterWrite(3 TimeUnit.SECONDS) .build(); public int eval(String pvid) { profileLog.error(\"RandomFunction invoked:\" + atomicInteger.incrementAndGet()); Integer result = cache.getIfPresent(pvid); if (null == result) { int tmp = (int)(Math.random() * 1000); cache.put(\"pvid\" tmp); return tmp;return resu @Override public void close() throws Exception { super.close(); cache.cleanUp();2. 单元测试