关于flink:Flink-SQL-在米哈游的平台建设和应用实践

摘要：本文整顿自米哈游大数据实时计算团队负责人张剑，在 FFA 行业案例专场的分享。本篇内容次要分为三个局部：

1. 倒退历程
2. 平台建设
3. 将来瞻望

点击查看直播回放 & 演讲 PPT

一、倒退历程

随着公司业务的倒退，实时计算需要应运而生。咱们依据重点的工作内容将倒退阶段划分为三个局部，

• 第一阶段是以 DataStream API 开发为主的 Flink 平台
• 第二个阶段是以 Flink SQL 为主一站式开发平台
• 第三阶段是一站式开发平台的性能深入和场景笼罩

第一阶段，以 DataStream API 开发为主的 Flink 平台，很好的解决了咱们对于实时计算的需要。但随着开发同学越来越多，大家发现基于 DataStream API 开发为主的实时计算平台，具备三个弊病，别离是开发成本高、版本易抵触、运维难度大，因而大家对 Flink SQL 的呼声就越来越高。

第二阶段，以 Flink SQL 为主一站式开发平台。次要的工作内容有：Flink SQL 能力晋升、指标和日志体系建设、元数据和血统治理。基于此，业务人员有了新的冀望。

• 第一，心愿平台可能更加智能化，升高用户的应用调参、调优等老本
• 第二，心愿流量的稳定可能具备主动扩缩容的资源管理能力
• 第三，心愿数据更具时效性。比方数据入仓、入湖后分钟级可查，或者基于近实时数仓开发

第三阶段，一站式开发平台性能深入和场景笼罩。次要的工作和将来要继续做的工作蕴含如下几个方面：

• 第一，工作资源的动态和动静调优能力
• 第二，资源的弹性扩缩容能力
• 第三，增强近实时数仓的建设

上面咱们进入平台的整体架构，从下图中能够看到平台总体蕴含三个局部，别离是用户权限及鉴权、性能和服务模块、以及环境和资源性能。

性能和服务次要蕴含作业大盘、概览、开发、运维、日志、元数据、血统、监控告警、资源调优、主动扩缩容、弹性资源管理以及平安管控等。

二、平台建设

那么基于这样的实时计算平台，咱们是如何建设的呢？围绕 Flink SQL 或者平台化的次要工作有如下四个方面：

• 第一，语义表白和控制能力的建设
• 第二，资源调优和弹性能力的建设
• 第三，指标体系建设
• 第四，近实时数仓建设

截止目前，Flink SQL 占比总任务数曾经在 90% 以上，极大的进步了大家的开发效率。上面咱们将对每一个局部进行具体的解说，来看一看具体都是怎么做的。

DataStream API 相较于 Flink SQL 有如下几个长处：

• 第一，算子并行度和传输方式可控
• 第二，执行图直观易于了解
• 第三，状态保留工夫能够别离设置。

但在转变到 SQL 的时候，会产生一些问题。基于此，咱们举个例子，来看一看为什么算子并行度和传输方式不可控了。

比方用户定义了一个 UDF 函数，用来解决 Kafka 数据源的某一个日志，而后将这个解决后的数据写入上游的 MySQL 或者其余存储。咱们假设 Kafka 某一个 Topic 分区有 10 个，整个工作的并行度设置为 20。这个时候就会发现，UDF 理论只会解决 10 个并行度的数据。Flink SQL 须要怎样才能拓展呢？

针对这种状况，咱们以后的解决方案是提供对执行图编辑的性能，依照编辑后果同 SQL 一起保留。如下图所示，有三个 Operator，Data Source Operator 的 ID=1，UDF Operator 的 ID=2，Data Sink Operator 的 ID=3。

在这个过程中，将整个作业的并行度设为 20，Source 源 Operator1 的并行度设置为 10，1 和 2 之间的传输方式设为 rescale。而后在后端接管到后，同步将 Job Graph 进行批改，就会失去如下的执行图，用户就可能比拟好的解决掉这个问题了。

对于这个问题，将来咱们的改良思路是通过 SQL 利用 Hint 性能来实现，或者更加智能化一点，依据作业指标信息，主动探测反压节点，自动化设置，来升高用户的应用老本。

对于 Create View 逻辑视图的含意是指什么呢？我也用一个案例来加以阐明。从下图能够看到，用户自定义了一个 UDF 函数模仿了一个数据源。咱们将这个数据进行解析，创立 Create View，比方叫 Row Table，而后向上游两个指标表 SinkTable1 和 SinkTable2 写入。最初看执行图，会发现 UDF 函数被执行了两次。

目前咱们针对这一个问题收集并提供了一些解决方案。但在提供解决方案之前，我想先论述一下这个问题产生的起因。Flink SQL 利用 Apache calcite 进行 SQL 语法解析，而后将解析后的 SQL 转换成一个语法树，通过 Flink Planner 生成 RealNode，通过 Optimizer Rule 进入 Codegen 环节。之后理论代码会有一个 Physical Plan 的过程，通过 Optimizer 造成 Steam Graph，而后转化成 Job Graph，最终转化成 Execution Graph。

那么 View 是在哪一层级失落的呢？其实是在 Apache calcite 语法解析的时候，View 它只是一个逻辑辅助，在这一过程会将其抛弃。那么咱们如何让 View 这一信息被底层感知到呢？

次要有两个方法:

• 方法一是 SQL 解析的时候不失落 View 信息
• 方法二是在 RealNode 到 Optimizer Rule 可能辨认到 View 的特色信息，这样就能够把 View 当成一个真正的代码去翻译了

方法一是一个十分好的解决办法，然而须要对 Apache calcite 进行很多改变，实现难度比拟大，老本也比拟高，所以采纳了方法二。最终的计划是采纳辨认特定函数实现，内置了一个 breakpoint 函数。在创立 View 的时候能够同时多 select 一个 breakpoint，这样在底层翻译的时候，就能够把它当成一个真正的 RealNode 解决。这个问题，将来咱们是也是心愿通过 SQL 利用 Hint 性能来实现。

对于状态的保留工夫方面咱们要怎么解决呢？以数据流关联 MySQL 分库分表的数据举例。常见的解决方案是利用 Flink CDC 将 MySQL 中的分库分表数据，抽取写入上游的 KV 存储中，而后再通过另一个 Flink SQL 工作接入 Kafka 关联，用时态表 Join 的形式将数据打宽，最终输入后果。

这一过程可能会有两个问题。第一，引入 HBase，咱们的工作就会从一个拆分成两个。其次须要假设上面这条链路的速度快于流的速度，否则下面 Topic 的数据达到的时候，而维表的数据还没达到就关联不上。那么怎么去解决这个问题，也是咱们思考的中央。

咱们采纳的计划是用 Flink SQL+CDC+Regular Join 的形式来实现。接入还是一样生产 Kafka，通过 CDC 来生产数据库分库分表的数据，最初通过失常的 Regular Join 来实现。

这里的 Regular join 底层同时依赖两个 MapState，比方 Topic A 对应 MapState 是 A，MySQL 里的数据库的数据对应的是 B。如果咱们能轻易的将 MapState B 的状态设置为 0 或者不过期，那么这个状态的数据就会被永恒的保留下来。即便流的数据先达到了，前面状态数据达到也能触发数据的关联，从而比拟好的解决这类问题。

具体的解决办法是，咱们能够在 Flink SQL 中指定左右流 Join 的状态工夫，在 Graph 中辨认有 Join 的算子，最终透传到 Join 算子做状态工夫的设置。

工作开发实现，须要多少资源呢？线上流量稳定，呈现提早怎么办？工作越来越多或工作并发调整，资源有余怎么办？

针对这些问题，咱们对应的解决办法次要蕴含：动态资源调优、动静资源调优及扩缩容、资源弹性能力的建设。那么具体咱们是怎么做的呢？上面请大家跟着我来一起来看一看。

举个例子，工作终于开发实现，通过了工作校验，然而工作参数，比方并行度、Slot、内存……该给多少能力失常运行呢？提供了如下三种 case：

• Case1：资源间接给足 –> 失常运行 –> 完结 —> 资源节约
• Case2：资源有余 –> 反压或者提早重大 –> 重复调整资源 –> 费时费力
• Case3：指标计算 Groupby–> 托管内存不足 / 增量 Checkpoint 没开 –> 工作运行一段时间失败

综上所述，三个案例的共性是工作调优老本高，且对用户自身有肯定的能力要求。对此咱们专门做了动态资源调优的解决办法。

假设用户开发了一个 Flink SQL，第一个环节，首先进行语法校验，而后通过语法校验及后端生成 Stream Graph，拿到 Stream Graph 的同时咱们还会进行 Source/Sink 连通校验和参数初步调整。

第二个环节，依据以后的工作逻辑及流量正当的调整资源。首先探测 Source 的流量，而后拿这个值和用户的作业 SQL、Stream Graph 做 Optimizer。Optimizer 局部次要包含 Restart、HighAvailable、Checkpoint、Parallelism、TaskManager、JobManager、StateBackend。

通过一直优化，失去一个比拟好的工作资源参数，供用户作为初始工作资源应用。如果探测的资源流量较大，Sink 到 MySQL 的 Batch 设置较小，针对这种状况，咱们会揭示 SQL 当中的参数进行调整，来帮忙用户更好的调整 SQL 工作的参数。

最终咱们会给用户提供给两个视图，别离是 SQL 自身调整的预览、工作所依赖参数的调整预览。如果用户感觉 ok，就能够依照以后的参数上线运行了。以上是动态资源调优。

那么工作上线后是什么状况呢？比方 Flink SQL 失常的 Running，首先将指标采集 Push 到 Kafka，而后会有实时工作进行指标的荡涤聚合。针对重要的指标，比方生产提早指标、算子速率指标、JVM 过程指标，状态大小指标等。

这些指标作为动静资源调整服务的入参，能及时感知到当前任务的运行状况，而后动静资源调整会进行需要资源的申请，将工作重启，并给用户发送告诉。如果重启失败，会进行配置回滚，而后告知用户调整失败须要手工染指。

针对动静资源调整，咱们的场景大略有如下四个：

• 设定历史数据追数：Kafka 积压历史数据首次生产、CDC 全量到增量。
• 冀望工夫动静调整：特定工夫扩缩容，解决流动可预知的流量顶峰。
• 依据指标动静调整：提早或反压及时调整，预测流量变动提前调整。
• 异样指标动静调整：例如 JVM GC 频繁，及时调整 TM 内存。

如上就是咱们想做的动静资源调优，最终实现的成果及具体的做法。

上面进入弹性资源能力的建设。过来咱们基于 Yarn On ECS 的形式，在扩容的时候须要较长的工夫。目前咱们基于 Yarn On K8s 来实现的，在 Yarn Label 上咱们会进行三种队列的设置打标签，固定资源队列对应的是正式工作；弹性资源队列对应的是突发流量工作；抢占资源队列对应的是测试工作。

如果忽然线上流量稳定，当前任务的固定资源有余。那么咱们就能够将通过分钟级的时效，将弹性资源队列资源扩进去，而后将任务调度下来。这样就防止了突发流量所带来额定资源的耗费，同时咱们也不须要依照最高峰值流量去预估资源，只需依照常定的工作资源数量来设定底层所须要的资源。

将来咱们将引进 Flink Native K8S，心愿借助 K8s 自身的资源管理能力提供资源弹性使用户有较好的体验。

指标体系在 Flink 工作中至关重要，次要蕴含工作可观测、动静资源调优和扩缩容、调度工作依赖三个方面。

• 第一，工作可观测方面，咱们的做法是采集指标到 Kafka，而后通过 Flink 荡涤聚合写入 Influxdb/MySQL，Grafana 展现 / 指标异样监控告警。
• 第二，动静资源调整和扩缩容的指标利用曾经后面阐明，就不再赘述了。
• 第三，调度工作依赖，是指 Kafka/MysqlCDC 数据入湖，上游有离线调度依赖，咱们须要感知当前任务是否有提早，Checkpoint 有没有做，数据在数仓里是否具备可见性，还须要保证数据残缺入仓入湖后，上游工作才会启动。

分享两个场景。第一个场景，日志场景建设。当数据量大，入仓工夫多于 10 分钟的时候，上游工作相应增大，有没有方法缩短入仓工夫？当 HDFS 写入流量稳定较大的时候，能不能更加安稳，且数据不丢不重？

家喻户晓，从日志文件通过 Kafka 到 Flink SQL、写入 Iceberg 都有可能产生数据反复，这一链路能保证数据不丢，但较难保证数据不重。

对此咱们的计划是基于文件日志采集 MetaData Logs，而后将 MetaData Logs 在上游复用。其中 MetaData Logs 的文件的行数起到很重要的作用，因为这一链路能保证数据不丢。

如果数据的行数等于 MetaData Logs，就代表这个数据没有反复，一旦数据行数多于 MetaData Logs，就代表这个数据有反复了，但咱们只须要基于反复的某一个文件日志进行去重解决，而不须要对全量日志文件都进行去重解决。基于这样解决形式，咱们发现入仓时效从原来的 10-20 分钟，升高到分钟级别的提早。同时这一链路也能保障入仓数据不丢不重，间接可用，等同于离线日志拉取 ETL 的场景。

针对 Iceberg 表咱们建设了 Iceberg Manager 来做小文件合并、过期快照清理、孤儿文件清理。

第二个场景，数据库场景建设。比方数据库是 MySQL，咱们想通过 Flink CDC 将数据间接写入 Iceberg V2 表。那么就会有如下几方面的思考：

• 多个 Flink CDC 工作是否会对一个 MySQL 读取？数据库是否会有压力？曾经读取的数据是否复用起来？
• Flink CDC 增量读取，反对指定读取的工夫终点。
• IcebergV2 全量数据同步时，数据量较大，容易产生了较多 Delete Files，辅助链路的 Iceberg Manager 在进行表级别优化的时候，就会产生较大的压力。
• Flink CDC 同步工作太麻烦，心愿配置化就生成好工作，心愿有一键数据入湖的能力。

基于此，咱们做了一个链路的辅助，一键工作生成。辅助主动工作的调优扩缩容机制，保障 Flink CDC 全量同步和增量同步资源的切换问题，通过 Kafka 来实现对同一个数据源读取时候的压力问题，将数据写入 Kafka，Kafka 的数据会被上游的 Flink SQL 工作主动感知并同步。

为了解决 Delete Files 全量数据过多的问题。咱们在进行全量同步的时候，会敞开写入 Iceberg V2 表的 upsert 性能，在增量的时候才会开启，这样就能够保障全量同步的时候数据既不丢也不重。同时，Flink SQL 工作增量数据会写入 Iceberg V1 表，不便上游链路进行复用。

三、将来瞻望

将来 Flink SQL 或者平台建设将围绕以下四个方面进行开展：

• 第一，批流一体。大数据离线数仓和实时数仓分为两套零碎，个别离线数仓通过 Spark、Hive 来实现，实时数仓应用 Flink。随着 Flink 批处理能力的一直建设，咱们认为应用一套批流一体，既能升高用户老本，还能更不便的防止两套引擎所带来的指标含意不同的影响。
• 第二，资源弹性能力的建设。将来会基于 K8s 一直引进弹性资源能力，更好的提供给用户应用。
• 第三，应用场景的建设，联合 Flink SQL 基于 Kafka 提供提早音讯的性能。
• 第四，近实时数仓 TableStore 的建设。TableStore 新版本公布，打算先实际起来，同时还将联合 Iceberg 一直摸索实际，实现让大家基于近实时数仓，就可能失去时效性和确定性两种交融的成果。

点击查看直播回放 & 演讲 PPT

更多内容

流动举荐

阿里云基于 Apache Flink 构建的企业级产品 - 实时计算 Flink 版现开启流动：
99 元试用 实时计算 Flink 版（包年包月、10CU）即有机会取得 Flink 独家定制卫衣；另包 3 个月及以上还有 85 折优惠！
理解流动详情：https://www.aliyun.com/produc…