为了全面推动数仓实时化，OPPO 基于 Flink 打造的实时计算平台 OStream ，已广泛服务于实时 ETL/实时报表/实时标签等应用场景。本文主要围绕基于Flink打造的实时计算平台OStream来展开，分享 OStream 平台的研发之道（包括设计原则、总体架构、Flink 改进优化），业务场景的接入与应用实践，以及平台往智能化方向发展的探索与思考。本文整理自OPPO大数据平台研发负责人张俊在Qcon全球软件开发大会 2019 广州站的演讲。如果读者正在考虑或者正在建设实时计算平台，希望能给大家带来一些参考。

OPPO大数据平台介绍

首先介绍一下OPPO的业务跟数据规模。OPPO是一家非常低调的公司，跟互联网、大数据到底有什么关系呢？简单地介绍一下。OPPO有自己的基于安卓的定制系统ColorOS，内置很多互联网应用，包括应用商店、浏览器、信息流等一些热门应用。这个系统经过几年的发展，日活现在已超过2亿。在业务驱动下，OPPO的数据量级从2013年开始到现在，几乎每年都是2到3倍的增长速度，所以现在总数据量规模也比较庞大，超过了100PB，每天新增超过200TB。

图1 OPPO业务与数据规模

在如此大数据量的规模下，不可避免要去构建数据的Pipeline。我们的主要数据来源是手机端的埋点数据，还有一部分是日志或者是DB的数据。我们基于NiFi开源的项目构建了整个接入系统，然后基于HDFS跟Hive做数据的存储跟计算。计算层主要有两部分，第一部分是小时级别的ETL，做数据清洗跟加工；然后还有一个部分是日级的Hive任务，做每天的汇总工作。整个Pipeline有一个调度系统，是基于开源ALflow做的定制版，我们称之为OFlow。

图2 数据处理Pipeline：离线模式

接下来是应用层。数据应用主要分三块：报表分析、用户画像、接口服务。也有一些自研产品。每天可以从Hive里把数据导到MySQL/Kylin、ES、Redis/HBase去支持产品的应用；还会基于Presto提供一些交互式查询等服务。这一整个基于离线批处理的Pipeline，经过2到3年的沉淀，几乎支撑了所有业务的发展。但是我们发现，最近几年整个互联网的人口红利不断消退，所以我们的业务被迫走向精细化运营的道路。

精细化的一个关键点是及时性，需要捕获用户即时行为跟短期兴趣。最典型的一个应用场景就是实时推荐，相信国内大部分公司都在做实时化推荐的事情。这样的及时性对整个数据处理就会有实时化的诉求。这里所说的实时化，是从过去小时级别、天级别向分钟级别跟秒级别过渡（现在还没有到微秒级别的诉求）。

刚才提到最重要的三块数据的应用——报表、标签、接口，分别列出了实际业务的一些场景。其实除了业务，还有一块容易被我们忽视的，是平台侧对实时化也有诉求。举个典型的例子，OPPO绝大部分离线调度任务都是在凌晨0点启动的。这个可以理解，因为是T+1的处理，到了凌晨0点就可以处理前一天的数据，所以大家都希望越早处理越好。但这对我们平台来说是一个很大的问题：一到凌晨，整个集群压力就非常大，经常半夜被电话叫起来，处理各种集群问题。对整个平台来说，如果能把批处理变成流式处理，把集中式的集群负载分摊到24小时，平台压力就会减少。

另外，如果有实时化流式处理，对整个标签导入，包括质量监控，也是很有帮助的。有了这么一个实时化的诉求之后，各个业务都在尝试搭建自己的实时流Pipeline。在搭建Pipeline时，技术选型最重要的两块，一是计算引擎，二是存储引擎。我们都知道，在开源的生态里，这两块现在是群雄争霸，有很多的选择。对于计算引擎来说，有老牌霸主Storm，还有Flink、Spark streaming这样的新贵；在存储引擎方面，有Redis、HBase、Elasticsearch等等。我们要构建一个Pipeline，计算引擎最终是要对接存储引擎的，很容易就会面临下图这样的局面，像一个很复杂的蜘蛛网。

图3 实时流处理的乱象

既然是一个数据平台团队，就有责任去构建一个平台收敛这些系统，因此会面临很现实的问题。比如如何说服业务，说现在就需要一个平台，需要你们接入我这个平台？怎么告诉他，我们这个平台是有价值的？站在平台角度，我们会认为这个价值是显而易见的，平台化能产生规模效应。规模效应带来什么呢？研发资源、硬件资源、运营资源的边际成本都可以降低。但反过来站在业务的角度，他不一定认同。他可能会觉得，你讲的是面，但他关注的是点，他会说：“我的业务快速发展是当下之急，你的规模效应确实是不错的，但对我业务来说，平台化会给我带来迁移的成本、学习的成本，另外还能给我带来怎样的价值？”

所以对于平台的建设来说，很重要的一点是：从一开始建设这个平台时，就需要站在业务赋能的视角看待这个问题。能不能屏蔽掉底层的一些细节，提高易用性？能不能提供更好的抽象，普适到更多用户？能不能跟业务签订服务等级协议（SLA），给他更好的服务保障？只有说，整个平台化从点到面地推进，先解决某个业务某个点的需求，慢慢把业务接进来，才能产生规模效应，最终，最大化平台的价值。这是我们在平台化推进当中的一个思考。

构建实时计算平台的技术实践

基于这样一个背景，实时计算平台如何推进呢？建设思路是怎样的呢？我们认为，不管构建系统也好，还是说构建一个平台也好，第一优先考虑是顶层设计。顶层设计什么呢？我们认为是两层，上层是API，下层是Runtime。“冰山理论”说：你所看到的东西，很可能只是它整体的很小一部分，事实上有更多你所看不到的部分隐藏在水面之下。这理论可以套用在很多地方。

对平台设计来说，我们希望平台的API尽可能简化，尽可能抽象，把更多的复杂性留在用户看不到的Runtime这一层。所以对API考虑是易用性、表达性和灵活性。到Runtime这一层，考核的是性能、健壮性和可扩展性。这些都是很复杂的分布式概念。

基于这么一个思路，平台API如何选择？既然说这个API面对用户，首先要考虑公司的人员分布与使用习惯是什么。

图4 我司人员使用习惯分布

我们在离线处理时代，几乎都是用Hive去做的，所以大部分人员都习惯于写SQL；可能还有一部分，习惯于写Scala、Java等，他们会希望通过Jar的方式提交数据处理；还有一部分，可能既不会写SQL，也不会写程序，会希望有一些简单易用的界面，直接做数据的分析。

可以发现，SQL是非常重要的平台一等公民。反过来我们再去考核一下SQL这个语言，大家都很熟悉，它是一个声明式语言，经过30、40年的发展，易用性、灵活性、表达性都很高，可以满足我们刚才提到的设计原则。

再到Runtime，我们的选择其实非常多。

图5 平台Runtime选择

这里仅仅列了几条，把核心诉求列了一个对比矩阵，发现只有Flink在各方面都满足需求。其他的引擎，像Spark Streaming，天然的micro-batch方式，没办法达到这种低延时的性能；对Kafka来说，它是比较轻量级的框架；对Storm来说，现在处于一个英雄垂暮的阶段，很多特性是在它那个时代是没有考虑过的。所以Flink相对来说是一个比较好的选择。

可能很多人会说现在Spark在发展，它从Structure streaming演进过来，也可以支持Continuous处理模式，可以支持低延时。但是我们认为，整个技术框架发展，技术最终肯定是趋同的。为什么选择Flink？还有一个很重要的原因是最近两年Flink在国内的发展普及程度。包括像阿里团队，他们也在大力地宣传跟投入，包括今天QCon大会上大沙和云邪两位老师，他们也是阿里团队社区的资深大V。

回过头来，Flink这个引擎最核心的优势是什么？哪些对我们来说觉得是一个亮点？首先看Flink引擎这一层，它最大的优势是Runtime，我们希望是高性能的，而它就是低延时、高吞吐。可能很多人会问，低延时跟高吞吐不是矛盾的概念吗？不应该去做一个权衡吗？其实Flink也没有去打破这个权衡，其实它内部有所谓的缓冲超时（buffer timeout）的机制，可以向左走，选择非常极端的低延时；也可以向右走，选择极端的高吞吐。那我们怎么看待它的所谓低延时、高吞吐呢？我是这么认为的：第一，不论是极端向左、向右走，最终的性能跟其他框架相比都是有一定优势的；第二，如果你的场景不需要非常极端走两边，可以直接站在中间，它也有比较综合的性能优势。大家可以看一下官网，有一些性能对比资料。

接下来就是端到端的“只说一次（exactly-once）”和高容错的状态管理，满足我们刚才所提到的健壮性要求。再接下来，对我们非常重要的是，基于事件时间和基于晚点数据处理的机制。因为对手机行业来说，手机上的埋点跟网页端埋点最大的区别是，数据的上报需要考虑用户的功耗和网络消耗。所以经常会出现用户的一次行为到这次数据真正的上报之间，有一个很随机的延迟，不能保证实时把数据报过来。所以基于事件时间去处理，包括处理晚点的数据，对我们来说，能够正确分析用户的行为，这很关键。最后，它还可以基于Yarn。因为我们在过去几年的发展中都是基于Yarn去做集群管理，所以我们希望能延续过去的经验。

我们再来看Flink SQL提供的能力。

图6 Flink SQL提供的能力

前面两条可以不用说，它是支持ANSI SQL跟UDF，包括数据类型、内置函数，这个在Flink官网有很详细的介绍。接下来，它可以自定义Source/Sink，这就满足了刚才说的可扩展性，我们可以不断扩充它的生态，如果它内置没有包含我们需要的上下游，可以自己做扩展。再接下来，就是之前其他讲师演讲里提到过的，像windows、join，还有批流统一的能力，这里就不复述了。

我们选择了平台的API跟Runtime之后会思考：整个平台从离线、批处理到实时的流处理，能不能是平滑的迁移？什么是平滑的迁移呢？前面有提到过，我们希望API这一层尽可能抽象跟简化，能不能在API这一层经过转变以后，对用户来说尽可能降低学习成本跟迁移的成本？我们发现这是可行的。所以我们会看到，在离线的时代，API这一层数仓抽象是table，编程接口是SQL+UDF；到了实时时代，我们其实可以保持一致，把更多的复杂性放在Runtime这一层，可以把计算框架、存储平台迁移到像Flink、Kafka这样的引擎上面，但是对于用户来说，API这一层看到基本的抽象是一致的，这样学习的成本是比较低的。

图7 离线到实时处理的平滑迁移

基于这么一个思路，最终推导出来整个数据Pipeline的实时模式，跟刚才我们说的离线时代几乎是一样的，只不过我们把一些关键性的组件替换了，比如说把HDFS替换成了Kafka，把Hive替换成了Flink，还有还有OStream平台，替换离线任务调度。然后其他的产品，包括基于报表、画像、接口这样的产品，都是保持一致的。

图8 实时模式与离线模式保持一致

有了这么一个建设思路以后，如何开展整个平台研发工作呢？首先做平台，就要把整体的架构画出来。然后大致分这么几个层次：最下面一层是基础引擎层，这一层大家都很熟悉，无非就是基于Kafka、Flink、ES等这些系统，整个集群管理通过Yarn来管理，然后Flink需要做检查点，会用到HDFS；平台特性这一层，提供网页版IDE，会有像元数据管理、流作业管理、日志检索、监控告警等这些能力；再往上就是API这一层，编程接口支持SQL，SQL是我们的第一等公民，然后还可以支持Jar包提交的方式；再往上，我们把它称为“集成工具”。刚才提到，我们会有一些场景，希望用户不用写SQL，可以通过简单UI配置的方式，就能自动生成SQL。然后我们还会跟内部的CI、CD工具自动打通，自动编译好Jar包之后，自动把Jar包传到我们这个平台，这个正在研发当中。这个就是整体的架构。

图9 OStream平台总体架构

前面我提到说SQL是整个平台的第一等公民，所以第一步要解决基于SQL开发的这么一个框架。刚才提到API这一层是基于Table SQL+UDF的方式，具体落地到我们的产品，我们需要有一个类似这样的界面（如果大家比较熟悉，这是开源的Hill的界面）：左侧有一系列的table，右边就是SQL的编辑框，可以提交SQL。这个开发界面，具体落到Runtime这一层需要怎么支撑呢？核心有两点：一个是元数据的管理，就是如何创建库表，如何上传UDF；还有一个就是流作业的管理，如何编辑SQL，如何提交作业，最终提交给Flink框架执行。

带着这个问题，我们看一下Flink SQL现在API的情况。

图10 Flink SQL API编程示例

这是最简单的Flink SQL编程示例，大概就是20行代码的样子。刚才提到流作业的管理，SQL的编译、提交，包括元数据的管理、上传table、注册table，都可以通过这个方式创建。但是我们还是不希望这样面向用户。因为用户不希望用编程的方式去实现。所以我们基于Flink简单地做了扩展。

整个过程大概是这样子：首先在我们的开发IDE上面，用户可以写一个SQL，写完以后可以提交，提交时创建类似于job的概念——我们会把SQL跟它所需要的资源、配置等封装起来，变成一个job，然后保存到MySQL。这个时候有一个job store，定期扫描MySQL，看看有哪些新的job，然后调用Flink里的TableEnvironment模块，真正去编译这个SQL，编译后会生成一个JobGraph（这是Flink认可的一个可执行的单元），最终再向Yarn去提交这个任务。

刚才提到过元数据管理——table创建出来以后，怎么被Flink识别？这里面涉及到编译过程中，通过元数据中心创建的table，如何被它识别到？我们实现了一个框架，用到了Flink里面的TableDescriptor概念，创建了一个TableDescriptor，即对表的描述符，也保存到MySQL里面。然后通过Flink的ExternalCatalog（刚才云邪老师也有提到过），可以把table识别出来，最终通过TableEnvironment去注册。这样，Flink就可以识别出外部的表，这个就是整个开发框架的实现。

图11 元数据中心的界面

左边是创建表的界面，我们可以创建像Kafka、MySQL、Druid、HDFS等的表。当然，我们其实不希望对用户暴露DDL，因为对我们的用户来说，他们也不希望去写create table，他们希望通过UI的方式创建表。右边是UDF的上传，可以自己写一个UDF，也可以把一个Jar包提交过来，然后指定你的入口类（main class）是什么。

图12 开发IDE界面

还有这个是我们的开发IDE，也是很简单的界面，其实就是模仿Hill的开发界面。

有了这么一个开发IDE以后，还要解决什么问题？真的要把它推向用户使用的话，还有两个基本的问题要解决。

一个是UDF，我们在Hive时代，积累了大量内部的UDF，包括加解密、格式转换，可能会有点问题；还有就是位置、距离的计算等等，其实有大量Hive的UDF。在Flink的处理上，大家也希望能够继承下来，直接使用它们。目前来说，Flink还不支持这一点。云邪老师刚才说，到1.9可能支持直接调用Hive的UDF，这个我们是很期待的，但目前来说还没有。我们作为平台方，需要把所有的UDF在Flink的框架上重新实现一遍。

第二就是维表的join。有一些业务的维表存在MySQL、HBase、Hive里面，我们需要去实现这一点。这在Flink1.9也将会实现，但现在的版本还没有，我们怎么做呢？

图13 维表创建页面

首先，这是我们的维表创建页面，可以创建一个MySQL的维表。由于大部分场景下，维表不算太大，可以直接把它cache到Flink的任务管理容器（task manage container）里面，所以我们这里可以选择ALL模式，即把所有的数据全量导入到容器。或者选择LRU（least recently used，最近最少使用）模式，表示这是可以刷新的。还有一个模式（NONE）是干脆什么都不做，每次都去查外部的数据源。

这是维表的实现，我们现在如何join呢？

图14 维表关联的实现

如上图，我们在平台层面会自动改写SQL。假设用户写了一个SQL，里面有一个join语法，我们在提交之前会加一层SQL的解析，解析出这是针对维表的join（因为我们知道哪个表是维表），把解析出来的上下文封装成一个JoinContext。有了JoinContext之后，我们会把SQL解析成更简单的SQL。我们把join的语法替换掉，替换成类似于另外一张表的形式。

背后怎么做呢？实际用到了Flink里可以把table跟stream做无缝转换的功能：把原始的table转成stream，然后在stream这一层再用flatmap的形式，每次去调用flatmap时，都会调用我们自定义的一个flatmap函数，这个函数会把刚才我们解析得到的JoinContext拿过来。这个JoinContext包含什么内容呢？包含维表到底是哪一张表、地址是哪些、关键字是什么。我们拿到这些信息之后，就可以去MySQL里面，在初始化阶段，open阶段，把维表全量加载到cache里，然后每次做flatmap时，就可以从cache里查找，最终通过stream跟table之间的转换，再转换回一个table，最终改写成这样的一个SQL。这就是我们当前的实现，看起来会有点奇怪，所以我们也很期待Flink原生对join的支持。

接下来，这个平台少不了的，就是日志检索。这也是平台化时需要提供的很常规的功能。我们现在可以对日志提供基于作业名称的、基于Yarn App ID的、基于container ID的全文检索。我们怎么做的呢？

图15 日志采集Pipeline

其实也比较简单，无论是SQL还是Jar包，通过OStream提交Flink作业时，我们会自动生成这样一个Log4j的properties配置文件，提供给Yarn以后，最终JobManager跟TaskManager在执行时，就会用到我们自定义的Log4j Appender，把我们的日志导向统一的Kafka，最终再把所有的日志导入ES做索引，就做到了全文索引。还有一个问题需要解决：所有的日志怎么跟OStream的job关联起来？比如需要知道每一行日志到底是属于哪一个job。这个比较好解决，我们在Flink里面有配置：在提交Flink作业时，可以给出6个环境变量，这6个环境变量分别注入给TaskManager跟JobManager，我们的Log4j Appender会自动识别进程里有没有这样的环境变量，如果有，把它写到ES里，每一行日志跟作业名称、App ID都可以关联起来。

接下来就是指标的监控，这也是平台里面需要提供的一个基本能力。有很多指标，但对于用户来说，最重要的两个指标是什么呢？一个是对Flink作业的吞吐，第二个是Kafka消费的lag。这个我们怎么做呢？

图16 指标采集Pipeline

首先，Flink的KafkaConsumer是一个源，对于每一个Flink job来说，KafkaConsumer会自动从Kafka的Consumer里继承它原生提供的很多指标，比如其中一个指标就是这个Consumer消费Kafka的position、消费的延迟是什么。对于所有的Flink Operators来说，也会暴露出来这么一个指标：这个Operator每秒出去了多少条数据。基于这两种指标，我们可以通过Flink原生的内部Matric的系统/子系统，比如它内部封装的，像MatricGroup、MatricRegistry，通过我们自己定制的一个MatricReporter，可以把我们刚才提到的两个指标导入到Kafka，最终也写到ES里去。然后基于这些指标做监控和告警。

图 17 告警规则

这是告警的页面，也是比较简单。基于这个指标可以做同、环比，或者是绝对值的告警。

上面大致介绍了基于Flink做的平台化的工作。这里做一个小结，是我们平台研发过程中积累的小小心得，简单分享一下。

第一个心得，对于我们这样的团队来说，最好的策略是维护纯净的分支。这个跟研发投入密不可分，最好的策略是紧跟社区的步伐，尽可能少改动Flink本身内核的东西。不然Flink整个社区发展很快，慢慢就追不上它的步伐了。怎么做呢？有两个策略。第一，任何的开源框架要打造它的生态，必然要去做很多扩展点，我们可以基于这些扩展点，做一些插件开发。比如刚才提到的元数据管理、如何把外部表对接到Flink里面去、如何做日志跟指标的采集，其实都是基于刚才提到的Catalog、Log4j Appender、Matric Reporter等等的扩展点。第二点是，如果没有扩展点可以开发插件，我们可以基于API去做二次抽象开发，不要改原生的代码。就像刚才提到的SQL的作业、维表的关联，其实都是基于原生的TableEnvironment、TableDescriptor这些API做的开发。

第二个心得，对于小团队来说，研发投入是有限的，大家如果想要去参与开源社区，怎么参加呢？我有些小的建议。首先我们要关注整个社区的动态。最好的方式有两个，第一个是可以关注一下Flip。每次大的变动在Flip都有介绍，从modification到design再到example，是一个完整的story，可以了解这个模块整个发展路径。还有一个就是关注它的pull request。这可能是最好的途径，因为每个想要往社区提交代码的，都要发一个pull request，可以关注到每个模块，比如像table、connect、runtime这些模块；可以看到作者们在内部的对话，看到社区的进展。第二个可以看作建立一个认知——如果你没有参与过开源的项目，可能对git branch的机制、怎么fork一个分支、怎么merge代码、怎么提交PR都没有认知。可以通过一些小的方式，比如说，可以去帮忙修正一些typo。你会发现Flink文档也好，注释也好，有很多语法错误，不管是语法拼写还是语法表达——这些就是typo。当然，提交这种（对于编程）没什么意义，但如果只是想建立一些基本认知，可以去做一些尝试。最后就是所谓快速调试，什么意思呢？你想去了解Flink某一个模块，或者是某一个想法想快速验证一下，不可能说连接外部的Yarn，然后再搭建一个Kafka，然后再往Kafka里插入一条数据，这样太慢了。最好的方式就是把端到端的集成化测试本地化起来，在自己的电脑上，不需要依赖任何外部数据源，可以直接跑通。还有，自动化起来：整个端到端，比如说往Kafka插入一条数据、往Yarn提交作业，都可以自动化起来，可以更好地了解Flink；甚至可以设一些断点，跟踪它的代码，了解Flink里面的内部机制。

这里给出了我自己的一个小工程，如果大家感兴趣可以了解一下。我利用很多像Kafka mini cluster、Yarn mini cluster的机制，在本地IDE可以把整个端到端的流程跑通，而且过程都是自动化的，不需要依赖任何外部的系统。

OStream平台运营实践经验

介绍了研发工作，最后再介绍一下平台运营方面的实践。第一个实践就是，我们认为最好的方式还是拆分离线跟实时集群。我们最开始的时候，所有的Flink都跑在离线集群当中。但我们认为是有问题的，为什么呢？整个离线集群跑的作业都是短平快的作业，资源分配很不确定。

图18 离线-实时计算集群拆分

如果大家对Yarn很熟悉的话，看到这张图应该会很亲切。我们可以看到，Yarn是公平调度机制，每个队列之间，资源分配都是不确定的。不确定性表现在哪里呢？大家可以看到上面有绿色的部分和橙色的部分。虽然可以给每个队列分配资源，但其他的资源随着每个队列的情况在不断动态变化。举个例子，比如说它有一个steady的fair share，这是固定分配的资源份额；还有instantaneous的fair share，表示如果说有一个队列，里面没有任何作业，那就要分享出来给别人去用。还有一块就是队列可以透支资源：比如有一些队列即使里面有程序在跑，但没有消耗完这个份额，这个份额需要被调出来给其他队列分享，其他队列可以透支。那这对我们来说是有问题的，线上集群出现好几次这样的问题。假设说批处理跟实时处理是两个不同的队列，可能我们在最开始的时候，两个队列是50%、50%的资源分配，假设实时作业有一些问题导致它要重启，重启的过程中，它的资源会立刻被另外一个队列抢占了，因为另外一个队列可以透支这个资源。所以对于线上队列，我们会把抢占给关闭，这可能是另外一个话题了。如果把抢占开启，其实也有很多的不确定性。

对我们来说，实时队列资源被抢占，导致重启之后就没资源跑起来。所以我们认为，虽然大家都在谈离线跟在线的混步，但我们没有精力研究混步到底怎么做，所以最稳妥的方式如果要提供服务的保障，最稳妥的方还是式把这两个集群做物理拆分。

还有测试，是从我们原先的实时处理模式系统继承过来的。大家知道，我们做数据开发，很重要的一点是，它的测试跟普通程序可能不太一样。很多时候，做数据开发，测试数据的逻辑，需要的不是随便造几条数据就可以了，需要的是全量的数据、生产的数据，不是随便造一些数据。我们以前的方式是这样的：在发布测试作业之前，读写的是测试数据库，而测试库的数据就从生产数据采样而来。但对业务来说，他们觉得这是有问题的。所以平台可以这样做：既然测试时需要依赖生产库，平台可以在提交SQL时做SQL改写，比如自动把insert into改为指向一个测试的库，但是读还是直接从生产库读，可以得到及时的、全量的数据，而不会对生产库造成任何的影响。这是我们的实践经验。

图19 全链路延迟监控

最后一个实践就是全链路延迟监控。上面这个链路是基于Flink的整个实时处理链路。数据从最开始的NiFi开始，经过三次Kafka，最终才真正到Druid里，然后才能从Druid里做报表。大家可以看到，在每个环节都有lag，从NiFi到Kafka，从Kafka到Aggregate，再从Kafka到Druid，有3次lag。之前做延迟监控都是针对单个作业，但用户关心的是他们的数据到这个系统之后，什么时候才能在报表中体现出来，关心的是全链路的延迟，单个延迟对他来说没有什么意义。所以我们需要构建血缘关系。

解析ETL、做Aggregate，把整个血缘关系解析出来，形成下面这个通路，从接入通道，到中间的table，再到中间的job，再到最终的Druid，这样我们可以把这4个lag加起来，变成一个统一的lag，做成一个统一的监控。大家可能还会有一个疑惑：中间ETL到Kafka、Aggregate到Kafka有两处空白，其实也有lag，为什么没有做监控？我们认为，Flink有比较好的反压机制。如果ETL到Kafka、Aggregate到Kafka有延迟，Flink通过反压机制，可以反映到上游的Kafka消费的lag里，所以我们可以忽略中间这两个lag，只计算那4个lag。

上面介绍了我们的研发和运营的工作，接下来分享一下案例。

图20 交互式查询界面

这个界面是以前离线时代提供的交互式查询界面。到了实时数据时代，用户觉得这样的界面还是挺好的，还是需要通过UI的方式、拖拽的方式直接分析实时数据。当然，我们会做一些限制：比如说通过拖拽的方式，不可能去查全量的实时数据，可能只能查最新的数据，比如最近一个小时或者几分钟的数据。

下面是实时的ETL，是我们线上应用最广泛的实时流处理应用。

图21 实时ETL

所有从手机端过来的数据，都会报到同一个通道里。在录入到数据仓库之前，需要对这些数据做一个拆分，我们用的就是Flink的SQL方式，比如我们可以写这4个SQL，对同一个table的数据，通过不同的条件，拆分到不同的下游表。我们用同样这一套处理逻辑，我们可以同时把数据插入到Kafka或者HDFS，作为后面离线处理的源数据。

还有就是实时标签。标签对我们来说是最重要的数据资产，所以实时标签是很重要的应用。

图22 实时标签

这个就是我们整个实时标签的通道，从Kafka里写一个类似于这样的SQL，插入一个表，格式做一定的限制——因为这最终要对接到我们的标签系统。大家也可以看到，我们用到了嵌套SQL，也用到UDF的函数（包括窗口）去实现。

未来展望和规划

最后做一些未来的展望：平台应该怎么走？

这可能说得比较大。我们认为一个平台发展的必然路径，肯定是要从自动化走向智能化，如果大家比较有信心的话，还会加一个词“智慧化”。大家对这三个名词的定义不太一样，尤其是“智能”，现在各种各样的地方都有智能这个词。

我这里给出我的定义。什么是自动化？我认为它是处理机械的、重复的事情，解放我们人的双手。最典型的自动化应用就是任务调度系统。每天到了一个点，重复性地、不需要人工干涉地自行调度任务，这是自动化。什么是智能化？一定是要做自适应跟自学习。因为不断自学习，所以它的行为应该是人没办法预料到的。可能最典型的应用就是一些AI系统，像阿尔法狗，是可以不断学习的。最后一个“智慧化”，可能比较虚，现实中可能没人见过。以前有部美剧《西部世界》，里面那些Hosts，就是有自我意识跟自我思维的，具备智慧化的。我们先抛开智慧化不谈，从自动化到智能化，听起来还是比较虚，具体落实到我们平台，我们能做什么事情呢？

比如说对自动化，我们可以做端到端的自动打通（我在后面再说）、通过规则自动生成SQL、每次提交作业时自动生成告警规则，这些是一些自动化的工作。什么是智能化？我们其实也可以尝试，比如说作业的资源它能不能自动伸缩，因为线上负载一定是曲线的，不是直线。如果说作业资源是直线，难免会有资源的浪费或者不够用的情况，能不能做自动伸缩。还有，作业出现异常能不能自我修复；作业参数这么多，不可能说在提交作业时考虑完善，有一个完美的参数配置，那么能不能在作业运行过程中自动动态调优。

首先看一下端到端的打通。这是一个真实的案例。

图23 以前SQL的Pipeline

这是我们以前SQL的Pipeline。数据处理完了，从Kafka读、往Kafka写；然后通过数据导入的方式导入到存储引擎，这可能是另外一拨维护存储引擎的同学做的；最终把它变成线上资产，可能是数据产品人员做的。这3个环节是割裂的，而且需要不同的人去处理。我们考虑能不能把这个过程自动化起来，不要给用户暴露出来说这是个Kafka table，而是就以展示表、标签表、接口表展现，这才是面向场景的。举例来说，对于展示表，无非有维度字段、指标字段、筛选字段、时间字段等。有这些字段以后，整个数据处理完，写完SQL以后，数据就可以自动加载到像Druid这样的平台里面去，最终也会自动变成一个报表，整个过程不需要用户再去操心。这就是线上正在做的例子。刚才也提到整个表的创建，未来表的创建是这样的：写了字段名称、描述、类型之后，可能还要判断，如果这是展示的表，则需要指定哪些是维度字段，哪些是指标字段。指定完成之后，当你写SQL，比如insert into，自动会跟下游像Druid、像报表系统打通，甚至可以尝试自动把报表创建出来，对用户来说，这就是端到端的流程。

最后分享一篇论文，是关于刚才说到智能化的。这是2017年微软跟Twitter发的一篇论文（他们做了一个叫Dhalion的系统），很有意思，它的关键字是self-regulating的，翻译过来可能是“自我监管”，满足我们刚才所提到的智能化的方向。

图24 线上作业智能化自动伸缩

这是从论文里摘出来的一张图，说明线上作业如何实现智能化自动伸缩。它是这么做的：首先是有一些Detector，像医生去诊断一样，去收集指标，根据指标判断是什么样的症状，比如说是产生反压了，还是线上的速率有倾斜了；获得症状后做一个诊断，通过不同的诊断器，判断到底是数据倾斜呢，还是有些实例因为磁盘变坏而变慢了；经过诊断以后，对症下药，如何解决这个问题。这可能也是自动的，有一些规则说怎么解决数据倾斜的问题、如何解决某个实例变慢的问题。

整个过程，包括中间的诊断，可以做成规则化，也可以做成基于机器学习，不断可以自适应的方式。这个框架基于Twitter之前做的Heron系统。我们能不能在Flink里面做这样的尝试？因为我们发现无论做指标收集也好，反压机制也好，到通过API对作业做自动伸缩，这些在Flink里都可以实现。我们能不能做这样一套框架，自动地、智能地在线上做资源的动态伸缩，这是挺有意思的一件事情。

创作场景

进击的实时数仓：Flink 在 OPPO 实时计算平台的研发与应用实践