1 大数据分析现状和背景

1.1 大数据分析现状

从 2008 年 Hadoop 成为 Apache 顶级项目之后，大数据技术经历了一个繁荣的发展阶段，各种组件层出不穷，上图显示了，当前查询分析软件有 300+，这也导致了当前大数据平台就像堆积木一样，数据在各个组件之间流转，需要冗长的 ETL 过程，数据存在多个副本，开发者需要多种系统的编程语言，开发难度高，使用复杂；

如上图所示，市场占有率最高也仅仅 11%，老牌数据库厂商 Oracle 仅仅占优 9%，而 others 占了 53%，即没有巨头，新进入者有巨大的机会；

从上图可以看出大数据的市场服务占比在 40%左右，而数据的服务占比在个位数，这里的服务指使用大数据的服务成本，包括开发，运维和维护，这也间接说明了大数据使用复杂的问题。

1.2 大数据分析面临的挑战

大数据当前面临的挑战归结起来有三点：

1. 数据经过 ETL 之后，存在多个副本，多个数据烟囱，管理复杂；

2. 引擎接口不同意，跨源关联分析的编程模型复杂，开发难度大；

3. 跨 DC 的数据访问，数据需要经过多次的数据搬迁，比如 DC1 访问 DC2，数据先要从数据源->DC2 的前置机->共享交互平台->DC1 的前置机->DC1 的数据分析引擎。

上述的挑战就要求分析引擎提供：（1）批/交互式融合分析；（2）跨源数据分析；（3）跨域协同分析。

2 openLooKeng 架构

openLooKeng 是一个统一高效的数据虚拟化融合分析引擎，我们的愿景是让大数据变简单，即要解决上述问题。 北向接口方面，openLooKeng 提供 ODBC、JDBC 以及 REST 接口，以 ANSI 2003 SQL 为载体提供统一数据访问接口，BI 工具、AI 工具可以有效地通过所提供的接口与 openLooKeng 集成，简化系统设计。南向接口方面，通过数据源连接框架，Data Source Connector 提供多种数据源的访问能力，无论是大数据生态的 Hive 或者 Hbase，或是 OLTP 数据库 PostgreSQL 以及 MySQL，都可以方便的接入。此外，openLooKeng 提供跨数据中心 Data Center Connector，提供高性能跨域协同计算。

openLooKeng 基础的交互式查询能力是基于 Presto 开源版本构筑的。但 openLooKeng 在技术场景、引擎内核技术、南北向应用生态等方面相对于 Presto 有较大差异。openLooKeng 是一个类 MPP 架构的分布式处理系统，包含协调器 Coordinator 以及 Worker 两种角色，通过实现 AA 高可用性，整体系统无单点故障问题。同时，openLooKeng 内部采用向量化列式处理，针对大数据场景列式处理性能更快，且可以充分利用 CPU 并行潜力。通过基于内存的流水线处理结构，openLooKeng 可以实现高性能并行处理。

同时，openLooKeng 提供以下三个特性：（1）高可用 Coordinator AA，防止单节点失效，（2）DC Connector 提供跨域分析的能力，（3）VDM（虚拟数据集市）简化数据的开发流程。

2.1 高可用 Coordinator

上图显示高可用 Coordinator 的架构，图中的 StateStore 是一个分布式缓存，当前使用的是 hazelcast，它有三作用：提供 lock 服务，用于选出提供 discovery 服务的 Coordinator（即主 Coordinator）；存储 discovery 信息；存储 query states。 Coordinator AA 的工作原理：首先，多个 Coordinators 在启动的时候，都会向 StateStore 申请 lock，谁先申请到，谁就是主 Coordinator，同时把自己的 ip 地址存储在 StateStore 里面；其次，worker 在启动的时候，从 StateStore 获取主 Coordinator 的 IP，之后的过程就和原始的过程一致；最后，如果主 Coordinator 挂了，则其他的 Coordinator 会感知，它们会去申请 lock，即抢主。

Coordinator AA 的作用：（1）多 Coordinator 同时运行并接收客户端的查询提交，（2）提供持续的应用可用性和抗灾能力，单个 Coordinator 故障不影响集群的正常运行，（3）高并发下，可减轻单 Coordinator 的压力，提高吞吐量，（4）结合 Nginx 等反向代理工具可实现负载均衡等高阶特性。

2.2 DC connector

DC connector 和 data source connector 的实现类似，只不过数据源是一个 DC。

DC coonector 的数据流如上图所示：

（1）获取分片，当前 DC 的分片是配置设置的一个值；

（2）DC1 调度分片；

（3）DC1 worker 收到分片处理请求之后，向 DC2 发送一个 Post 请求，实际上是一个 SQL 请求；

（4）DC1 的 worker 一直调用 Http Get 从 DC2 获取数据，直到数据获取完成。

通过 DC connector 打通异地数据中心数据访问，跨 DC 数据协同查询无需依赖数据中转平台，且通过算子下推与跨域动态过滤技术，可获得广域网部署，局域网的性能体验。

2.3 虚拟数据集市 VDM

通过 VDM，可方便的对底层的数据源、数据表进行管理，通过建立轻量级的视图来实现对不同数据源的模式化访问，使得用户不需要每次查询都关心数据的分布以及访问方式，从而简化数据开发过程。

3 性能优化实践

3.1 优化技巧全景

优化的技巧主要包括：

（1）在数据源侧，更适应 openLooKeng，针对 Hive 数据源分桶/分区、小文件合并、查询字段排序等策略可以让数据源更加适配 openLooKeng，提升整体性能。

（2）引擎层，增强交互式查询能力，包括多种缓存加速（执行计划缓存、元数据缓存、增量列式缓存）。

（3）增强优化器，包括谓词下推、动态过滤、RBO&CBO 等提升性能。

（4）采用自适应调度器，让数据处理更贴近数据源。

（5）额外层，加速交互式查询，包括启发式索引 Heuristic index layer（bitmap/bloomfilter/min-max）、Data cache layer 同时提供序列化 &反序列化优化。

3.2 稀疏索引

当前，Presto 在分区列上已经可以进行分区过滤，但是非分区上不可以。为了应对上图中的 sql，openLooKeng 提供稀疏索引来进行分区裁剪。

Bloom filter 索引，确定每个 split 是否包含要搜索的值，并只对可能包含该值的 split 进行读操作：（1）可以快速判断一个集合中有无某个值；（2）需要预先通过 create index 进行索引创建，openLooKeng 提供类似于 DB 的 create index 命令，可以创建 bloom filter、min/max 和 bitmap 索引；（3）通过在 coordinator 侧过滤，减少不必要的 split 生成与处理。

上图是真实客户场景的测试结果，客户的场景分析如下： （1）单表数据量很大，只有天分区，测试数据包含 30 天；（2）谓词包含 OR 以及 AND；（3）单表点查询，聚合操作，无 join。客户的要求是 100 并发下，30 天的数据查询在 3 秒内返回。

在没有创建索引前，单并发的查询性能在 8~10 秒左右。通过添加索引，可以看出：只 50 并发下都满足了需要，30 天 100 并发的查询还存在一定差距。后续的优化思路包括：索引 OR 支持，并且下推 OR 操作；聚合场景的 Aggregation Stage Cache 和 StarTree Index。

3.3 动态过滤

TPC-DS 是 TPC 标准组织推出的一个广泛使用的行业标准决策支持基准，用于评估数据处理引擎的性能。在 TPC-DS 是一个典型的星型 &雪花型的数据仓库组织方式，包含 7 张事实表和 17 张维度表，事实表数据量极大，而维度表相对较小，查询很少有谓词直接应用到事实表，事实表查询条件通过维度表相连接得到。

问题：传统谓词下推等优化很难应用，因为 probe 侧的表无法做到有效过滤，几乎是全表进行扫描参与 join，导致 join 数据量巨大导致执行时间过长。

针对这种场景，openLooKeng 采用动态过滤（Dynamic Filtering）技术。依靠 join 条件以及 build 侧表读出的数据，运行时生成动态过滤条件（dynamic filters），应用到 probe 侧表的 table scan 阶段，从而减少参与 join 操作的数据量，有效地减少 IO 读取与网络传输。

整体来看，通过在 build 侧表的 table scan 之上添加 DynamicFilterSource 算子，搜集 build 侧数据，通过分布式缓存进行 DF 的处理，最终经过 coordinator 端 DynamicFilterService 的合并，生成最终可以应用的条件，推送给 probe 侧的 table scan 进行数据过滤。

TPC-DS 的性能结果如上图所示，开启动态过滤，TPC-DS 测试用例的执行时间减少了 38.9%。

3.4 全局动态过滤

为了提高跨 DC 的访问性能，openLooKeng 把动态过滤应用到跨 DC 的数据访问，即全局动态过滤特性。设计思路如下：

DC2 coordinator 侧：

（1）增加业务用户的生产能力并提升效率；

（2）通过 session 判断当前 query 是否存在 cross-region-dynamic-filter，判断依据是 dc-1 的 DC Connector 在下推子语句到 dc-2 时，会监测是否有 dynamicFilter，若存在，则会在 HTTP Request 中设置属性标签，在 dc-2 的 coordinator 接收语句时，将 session 中的设置启用 cross-region-dynamic-filter；

（3）coordinator 完成子语句的 Analyze 会生成 Plan，从 Plan 中提取子语句的列名到 Plan 中各个 PlanNode 的 outputSymbol 的映射关系，同时，在这个过程中，判断 TableScanNode 是否 DC table，若是，则标记下来，可能存在需要继续下推 filter 的可能；将 mapping 和 DC 标记记录存入到 hazelcast。

DC2 worker 侧：

（1）CrossRegionDynamicFilterOperator 从 hazelcast 根据 queryId 获取 filter，通过 OutputNode 中 columnNames 和 outputSymbols 的映射判断是否存在需要过滤的列，若存在，则对 Page 中的该列进行过滤；

（2）对从 Connector 获取的 Page 数据进行过滤；涉及的 Operator 有三个，TableScanOperator、ScanFilterAndProjectOperator、TableScanWorkerProcessorOperator; 从 hazelcast 中根据 queryId 拉取 filter，mapping，DC 标记记录，然后根据规则对 Page 的对应列进行 filter，以及生成新的 bloomFilter 并存入 hazelcast；

（3）DC Connector 从 hazelcast 获取是否需要下推到远端集群的 filter。

性能测试如下：

SQL-2 性能提升这么明显于以下几个原因：

（1）通过动态过滤，DC2 右表的 TableScan 操作之后返回给上一阶段的数据明显降低了几个数量级，从而极大的减少了数据 Join 的时长；

（2）减少了 DC2 到 DC1 的数据传输；

（3）在 DC1 上进行 join 的数据量极大减少了。

3.5 TPC-DS 性能测试

优化技巧总结：（1）动态过滤：q64，q78，q75，q40，q49（2）Semi join 转 inner join：q95，q14，q93，q58（3）window function + filter 转 Top(N+M)：q67（4）使用 group by 消除 self join: q95（5）join reorder: q64

4 总结

1. 统一北向数据访问接口，丰富南向数据源，实现跨数据源数据免搬迁融合分析。

2. DC connector 提供跨域分析能力，并且通过全局动态过滤，算子下推，压缩断点续传，Coordinator AA 等技术来提高 DC connector 的性能和稳定性。

3. 通过元数据 cache，执行计划优化，索引，动态过滤，算子下推等特性，整体提高 openLooKeng 的性能。

• • •

openLooKeng 是一款开源的高性能数据虚拟化引擎，提供统一 SQL 接口，具备跨数据源/数据中心分析能力，为大数据用户提供极简的数据分析体验。欢迎加入 openLooKeng 社区，一起做点有意思的事儿，让大数据更简单！

openLooKeng 开源社区官方网站:

https://openlookeng.io/zh-cn/

openLooKeng 代码仓地址:

https://gitee.com/openlookeng