摘要：本文由阿里巴巴技术专家贺小令（晓令）分享，主要介绍 Apache Flink 新场景 OLAP 引擎，内容分为以下四部分：
- 背景介绍
- Flink OLAP 引擎
- 案例介绍
- 未来计划
一、背景介绍
1.OLAP 及其分类
![640.png]( "640.png")
OLAP 是一种让用户可以用从不同视角方便快捷的分析数据的计算方法。主流的 OLAP 可以分为3类：多维 OLAP ( Multi-dimensional OLAP )、关系型 OLAP ( Relational OLAP ) 和混合 OLAP ( Hybrid OLAP ) 三大类。
（1）多维 OLAP ( MOLAP )
传统的 OLAP 分析方式
数据存储在多维数据集中
（2）关系型 OLAP ( ROLAP )
以关系数据库为核心，以关系型结构进行多维数据的表示
通过 SQL 的 where 条件以呈现传统 OLAP 的切片、切块功能
（3）混合 OLAP ( HOLAP )
将 MOLAP 和 ROLPA 的优势结合起来，以获得更快的性能
以下将详细介绍每种分类的具体特征。
■ 多维 OLAP ( MOLAP )
MOLAP 的典型代表是 Kylin 和 Druid。
![640 1.png]( "640 1.png")
首先，对原始数据做数据预处理；然后，将预处理后的数据存至数据仓库，用户的请求通过 OLAP server 即可查询数据仓库中的数据。
![640 2.png]( "640 2.png")
MOLAP 的优点和缺点都来自于其数据预处理 ( pre-processing ) 环节。数据预处理，将原始数据按照指定的计算规则预先做聚合计算，这样避免了查询过程中出现大量的临时计算，提升了查询性能，同时也为很多复杂的计算提供了支持。
但是这样的预聚合处理，需要预先定义维度，会限制后期数据查询的灵活性；如果查询工作涉及新的指标，需要重新增加预处理流程，损失了灵活度，存储成本也很高；同时，这种方式不支持明细数据的查询。
因此，MOLAP 适用于对性能要求非常高的场景。
■ 关系型 OLAP ( ROLAP )
ROLAP 的典型代表是 Presto 和 Impala。
![640 3.png]( "640 3.png")
ROLAP 的处理流程上，用户的请求直接发送给 OLAP server，然后 OLAP server 将用户的请求转换成关系型操作算子，再通过 SCAN 扫描原始数据，在原始数据基础上做过滤、聚合、关联等处理，最后将计算结果返回给用户。
![640 4.png]( "640 4.png")
ROLAP 不需要进行数据预处理 ( pre-processing )，因此查询灵活，可扩展性好。这类引擎使用 MPP 架构 ( 与Hadoop相似的大型并行处理架构，可以通过扩大并发来增加计算资源 )，可以高效处理大量数据。
但是当数据量较大或 query 较为复杂时，查询性能也无法像 MOLAP 那样稳定。所有计算都是临时发生 ( 没有预处理 )，因此会耗费更多的计算资源。
因此，ROLAP 适用于对查询灵活性高的场景。
■ 混合 OLAP ( HOLAP )
混合 OLAP，是 MOLAP 和 ROLAP 的一种融合。当查询聚合性数据的时候，使用MOLAP 技术；当查询明细数据时，使用 ROLAP 技术。在给定使用场景的前提下，以达到查询性能的最优化。
2.Apache Flink 介绍
■ Flink 支持的应用场景
![640 5.png]( "640 5.png")
Apache Flink 支持的3种典型应用场景：
（1）事件驱动的应用
（2）流式 Pipeline
（3）批处理 & 流处理分析
■ Flink 架构及优势
![640 6.png]( "640 6.png")
Flink 的整体架构如上图所示，在此架构下，Flink 的优势也十分突出，主要分为6个方面：
（1）统一框架 ( 不区分流处理和批处理 )
（2）多层次 API
- 标准 SQL APL
- Table API
- DataStream API ( 灵活，无 schema 限制 )
（3）高性能
（4）方便集成
- 支持丰富的 Connectors
- 方便对接现有 Catalog
（5）灵活的 Failover 策略
- 在 Pipeline 下支持快速 failover
- 类似 MapReduce、Spark 一样支持 shuffle 数据落盘
（6）易部署维护
二、Apache Flink OLAP 引擎
1.为什么 Flink 可以做 ROLAP 引擎？
![640 7.png]( "640 7.png")
- Flink 的核心和基础是流计算，支持高性能、低延迟的大规模计算。
- Blink 将批看作有限流，批处理是针对有限数据集的优化，因此批处理引擎也是构建在流引擎上 ( 已开源 )。
- OLAP 是响应时间要求更短的批处理，因此 OLAP 可以看作是一种特殊的批。OLAP 引擎也可以构建在现有的批引擎上。
注：Flink OLAP 引擎目前不带存储，只是一个计算框架。
2.Flink 做 OLAP 引擎的优势
![640 8.png]( "640 8.png")
（1）统一引擎：流处理、批处理、OLAP 统一使用 Flink 引擎。
降低学习成本，仅需要学习一个引擎
提高开发效率，很多 SQL 是流批通用
提高维护效率，可以更集中维护好一个引擎
（2）既有优势：利用 Flink 已有的很多特性，使 OLAP 使用场景更为广泛。
使用流处理的内存计算、Pipeline
支持代码动态生成
也可以支持批处理数据落盘能力
（3）相互增强：OLAP 能享有现有引擎的优势，同时也能增强引擎能力
- 无统计信息场景的优化
- 开发更高效的算子
- 使 Flink 同时兼备流、批、OLAP 处理的能力，成为更通用的框架
3.性能优化
OLAP 对查询时间非常敏感，当前很多组件的性能不满足要求，因此我们对 Flink 做了很多相关优化。
■ 服务架构的优化
下图介绍了一条 SQL 怎么在客户端一步一步变为 JobGraph，最终提交给 JM：
![640 9.png]( "640 9.png")
在改动之前，每次接受一个 query 时会启动一个新的 JVM 进程来进行作业的编译。其中 JVM 的启动、Class 的加载、代码的动态编译 ( 如 Optimizer 模块由于需要通过 Janino 动态编译进行 cost 计算 ) 等操作都非常耗时 ( 需要约3~5s )。因此，我们将客户端进行服务化，将整个 Client 做成 Service，当接收到用户的 query 时，无需重复各项加载工作，可将延时降低至 100ms 左右。
- 自定义 CollectionTableSink
![640 10.png]( "640 10.png")
这部分优化，源于 OLAP 的一个特性：OLAP 会将最终计算结果发给客户端，通过JobManager 转发给 Client。假如某个 query 的结果数据量很大，会让 JobManager OOM ( OutOfMemory )；如果同时执行多个 query，也会相互影响。
因此，我们从新实现了一个 CollectionTableSink，限制数据的条数和数据大小，避免出现 OOM，保证多个 Query 同时运行时的稳定性。
![640 11.png]( "640 11.png")
在 Batch 模式下的调度存在以下问题：
- 使用 Lazy_from_sources 模式调度，会导致整体运行时间较长，也可能造成死锁。
- RM ( Resource Manager ) 按 OnDemand 方式分配 Slot 需求，也会造成死锁。
- RM 以单线程同步模式向 TM ( Transaction Manager ) 分配 Slot 请求，会造成等待时间更长。
注：调度死锁是指在资源有限的情况下，多个 Job 同时运行时，如果多个 Job都只申请到了部分资源并没有剩余资源可以申请，导致 Job 没法继续执行，新的 Job 也没法提交。
针对上述问题，我们提出了以下几点改动：
- 采用 Eager 调度模式 ( 确保所有的资源都申请到后才开始运行 )。
- 使用 FIFO ( 先进先出队 ) 模式申请资源 ( 确保当前 Job 的资源分配结束后才开始下一个 Job 的资源分配 )。
- 将单线程同步模式改为多线程异步模式，减少任务启动时间和执行时间。
■ 针对 source 的优化
在 ROLAP 的执行场景中，所有数据都是通过扫描原始数据表后进行处理；因此，基于 Source 的读取性能非常关键，直接影响 Job 的执行效率。
![640 12.png]( "640 12.png")
像 Parquet 这类的列存文件格式，支持按需读取相所需列，同时支持 RowGroup 级别的过滤。利用该特性，可以将 Project 和 Filter 下推到 TableSource，从而只需要扫描 Query 中涉及的字段和满足条件的 RowGroup，大大提升读取效率。
![640 13.png]( "640 13.png")
这个优化也是充分利用了 TableSource 的特性：例如 Parquet 文件的 metadata 中已经存储了每个 RowGroup 的统计信息 ( 如 max、min等 )，因此在做 max、min 这类聚合统计时，可直接读取 metadata 信息，而不需要先读取所有原始数据再计算。
■ 在没有统计信息场景下做的优化
![640 14.png]( "640 14.png")
CrossJoin 是没有任何 Join 条件，将 Join 的两张表的数据做笛卡尔积，导致 Join 的结果膨胀非常厉害，这类 Join 应该尽量避免。我们对含有 CrossJoin 的 Plan 进行改写：将有 join 条件的表格先做 join ( 通常会因为一些数据 Join 不上而减少数据 )，从而提高执行效率。这是一个确定性的改写，即使在没有统计信息的情况下，也可以使用该优化。
![640 15.png]( "640 15.png")
通常情况下，两阶段的 Aggregate 是非常高效的，因为 LocalAggregate 能聚合大量数据，导致 Shuffle 的数据量会变少。但是当 LocalAggregate 的聚合度很低的时候， Local 聚合操作的意义不大，反而会浪费 CPU。
在没有任何统计信息的情况下，优化器没法决定是否要产生 LocalAggregate 算子；因此，我们采用运行时采样的方式来判断聚合度，如果聚合度低于设定的阈值，我们将关闭聚合操作，改为仅做数据转发；经我们测试，部分场景有 30% 的性能提升。
4.测试结果
![640 16.png]( "640 16.png")
上图是 Flink 和 Presto基于 1T 数据做的 SSB ( Star Schema Benchmark ) 测试，从图中可以看出 Flink 和 Presto 整体上不相上下，甚至有些 Query Flink 性能优于Presto。
注：Flink OLAP 从开始到嘉宾分享时，只有3个月时间。
案例介绍
1.Flink OLAP 在数据探查上的应用
![640 17.png]( "640 17.png")
上图描述了一个数据湖应用的完整架构，Flink OLAP 主要用于"数据探查"。
数据探查是对数据结构做智能判断，给出数据的探查结果，快速了解数据的信息和质量情况。即用户可以在管控平台上了解数据湖中任意一份数据的数据特性。用户通过 Web 交互操作选择相应的表和指标后立即展示相关结果指标，因此要求低延迟、实时反馈。而且数据湖中很多数据没有任何统计信息；前述的各种查询、聚合层面的优化，主要为这类场景服务。
2.整体架构
![640 18.png]( "640 18.png")
上图是这类应用的整体架构。整套服务托管到 Kubernetes 上，最终访问的数据是OSS。
未来计划
当前，Flink OLAP 引擎性能优化及应用主要是基于内部 Flink，后续工作主要分为以下三块：
- 推回社区：目前所有工作都是基于内部 Flink，希望推回社区；
- 资源隔离：后期很多功能的开发和优化会围绕多 Query 运行时的"资源隔离"；
- 优化&性能：围绕 OLAP 的特性，在此场景下会进一步做优化和性能提升等方面的工作。
