ClickHouse 在百度 MEG 数据中台的落地和优化
导读
百度 MEG 上一代大数据产品存在平台分散、质量不均和易用性差等问题，导致开发效率低下、学习成本高，业务需求响应迟缓。为了解决这些问题，百度 MEG 内部开发了图灵 3.0 生态系统，包括 Turing Data Engine(TDE)计算引擎、Turing Data Studio(TDS)数据开发治理平台和 Turing Data Analysis(TDA)可视化 BI 产品。依托图灵 3.0 生态，我们进而形成了一套新的开发范式——"OneData+开发范式"，其关键在于可视化分析与数据集的构建。
TDE-ClickHouse 作为图灵 3.0 生态中重要的基础引擎之一，专注于为业务提供海量数据下的自助秒级分析能力。通过高性能的数据查询能力与高效的数据导入通路，支持业务更及时、敏捷地对海量数据进行分析；通过稳定可靠的分布式架构，在减少资源和运维成本的同时，严控引擎侧的数据质量。
01 百度 MEG 数据中台解决方案
1.1 背景与问题
上一期的 Geek 说我们分享了图灵 3.0 中的数据开发治理平台 TDS（Turing Data Studio），这一期我们分享图灵 3.0 生态中的一个重要的基础引擎 TDE-ClickHouse。首先我们再回顾一下图灵 3.0 生态的发展背景：
百度 MEG 上一代大数据产品存在平台多、质量参差不齐和易用性差的问题。这些问题导致开发人员面临较高的研发依赖、开发效率低下和高昂的学习成本；业务部门则感知需求支持迟缓、数据产出延迟及数据质量低的问题。
1.2 图灵 3.0 生态概述
为解决上述问题，我们构建了新一代大数据解决方案——"图灵 3.0"，旨在覆盖数据全生命周期，支持全链路数据操作，提供高效敏捷且统一的强大数据生态系统，其中包括数据计算引擎、数据开发和数据分析三个核心部分：
（1）TDE（Turing Data Engine）：图灵生态的计算引擎，包括 ClickHouse 和 Spark；
（2）TDS（Turing Data Studio）：一站式数据开发治理平台；
（3）TDA（Turing Data Analysis）：新一代可视化 BI 分析产品。
△图灵 3.0 生态产品
TDA 以可视化分析为主的产品形态，要求底层引擎需要具备秒级响应的能力。CH 凭借列式存储、数据压缩、向量化执行等特性，以及对硬件资源的极致使用，在查询性能方面表现出众。因此在可视化分析场景我们选用 CH 作为核心计算引擎，对于超大数据量的查询及 ETL 加工场景则通过 Spark 支持。
1.3 OneDate+开发范式
依托图灵 3.0 生态产品体系，我们进而形成了一套新的开发范式——"OneData+ 开发范式"，其中关键在于可视化分析与数据集的构建，为业务提供海量数据下的自助秒级分析能力。
传统的分层数仓建模方法将数据分为 4 层，包括 ODS（Operational Data Store）、DWD（Data Warehouse Detail）、DWS（Data Warehouse Summary）和 ADS（Application Data Store）。在这种建模方法中，数据开发侧作为需求的被动承接方，根据业务侧提出的数据需求进行数据开发与交付，存在需求交付周期长、人力成本高等问题。
OneData+开发范式引入"数据集"的概念，其中数据开发侧根据具化的业务主题与领域知识主动建立对应数据集，并支持数据集的后续迭代，业务侧则基于数据集进行交互式的自助拖拽分析，极大的提升了数据开发效率，降低了数据运维成本。
△OneData+开发范式
1.4 ClikHouse 的挑战
在 OneData+开发范式中，数据开发侧与业务侧的依赖关系从之前的"需求-交付"解耦为以数据集为中心的建设。在实践中，大部分数据集同之前 DWS 层的表类似，会结合业务场景建设得更加成体系，保证数据需求尽量不重不漏。ADS 层则彻底消失，无需数据 RD 单独开发。这种情况下 DWS 层数据相对于 ADS 层数据膨胀了好几个数量级，但是要求查询性能不明显低于之前直接查询 ADS 层数据，这样就对 CH 的查询性能提出了极高的要求。
百度 MEG 业务实践过程中，对 CH 的要求是：保证百亿级的数据分析请求能秒级响应。因此，为了提升 CH 的查询性能，我们做了大量的优化工作。
此外，随着 CH 在内部的大规模使用，在数据导入和分布式架构方面也暴露出一些不足之处，比如：
（1）大规模数据导入难以保证导入任务的稳定性以及时效性，甚至产生级联的资源抢占问题；
（2）CH 分布式架构存在运维成本高等问题，并且缺乏集群层面的事务支持等高级功能。
这篇文章将从查询性能、数据导入和分布式架构三个方面介绍我们对 CH 所做的优化工作。
02 ClickHouse 查询性能优化
2.1 计算机分层解耦，聚合层硬件加速
原生 CH 架构主要采用单层同质的分布式架构，集群中所有数据节点的资源配置相同，数据按照分片划分并做副本冗余，存储在各个节点的本地磁盘中。
查询客户端会随机选择一个数据节点作为协调者节点，并发起查询请求。相比其他数据节点，协调者节点除了本地子查询的计算外，协调者节点额外负责:
（1）初始查询拆解成子查询后下发；
（2）合并各个子查询的结果后返回。
协调者节点需要额外的网络 IO 来获取其他节点的结果，以及更多的 CPU 来合并结果。在大数据量和复杂查询场景下，网络 IO 和合并结果会占用大部分时间。因此在理想状态下，协调者节点需要更高的 CPU 和网络配置，但在 CH 原生的单层同质架构下，让所有节点都具备高规格资源配置会导致成本明显上涨并出现资源浪费。
因此我们将协调者额外的职责抽离出来，独立成一层"聚合层”。聚合层节点拥有更高的 CPU、内存和网络吞吐能力，其本身不存储数据。所有客户端查询都会先路由到聚合层，再由聚合层作为协调者与 CH 集群的数据节点交互。同时聚合层的存在让跨物理 CH 集群的查询（比如跨集群 Join）变为可能。
△计算分层解耦
2.2 数据多级聚合
数据开发侧首先基于 OneData+开发范式在数据建模层面进行数据聚合，基于大宽表（事实表 + 维度表），并结合业务主题抽取出相应的维度聚合表作为 CH 数据集。为满足业务侧的拖拽式分析，CH 数据集多为细粒度主题宽表（单表百+字段，单天数亿行数据）。
而在实际查询场景中，由于大部分查询请求都是报表类型的查询（单次查询涉及平均个位数字段，返回结果集数百行数据），因此需要在保证查询灵活性的同时，不能影响报表类查询的性能，此时需要在 CH 计算引擎层面进行透明的数据聚合。
计算引擎层我们实现了两层数据预聚合：
（1）引入聚合表：基于 Projection 实现对查询中间状态的预聚合，避免对原始明细数据的大量磁盘扫描；
（2）缓存查询结果：将最终查询结果缓存在外部内存，缩短查询链路，并避免重复查询带来的多余磁盘 IO。
△数据多级聚合
2.2.1 自动 Projection
经过对各类聚合表的选型对比，我们最终选择 Projection，因为和传统聚合表相比，Projection 有两点优势：
（1）Projection 与底表数据强一致: 如果 Projection 中有数据缺失，查询会自动透传到底表；
（2）对客户端完全透明：无需修改业务 SQL，引擎会自动判断是否命中 Projection。
△ClickHouse Projection
同时我们与下游 BI 平台打通，针对 BI 平台中的高频查询自动创建 Projection，并对 Projection 进行全生命周期自动化管理，具体包括：
Projection 识别
解析查询 SQL，识别其中的查询字段（包括维度与指标字段）与计算函数；
基于成本评估聚合效果，过滤低效的 Projection，并结合其他拦截策略（比如集群负载等）做服务降级；
生成待创建的 Projection 语句。
Projection 生命周期管理
Projection 创建主要基于生成的 Projection 语句并在 CH 集群创建；
Projection 删除主要关注命中率较低的 Projection 的挖掘以及回收；
Projection 物化主要关注数据上卷在历史存量数据中生效。
△自动 Projection 建设
2.2.2 查询结果 Cache
对于查询结果的缓存，原生 CH 的 QueryCache 存在如下问题：
（1）单机 Cache，多个“聚合层”实例的 Cache 无法共享；
（2）只有基于时间的过期机制，无一致性保证，底层数据更新后可能查到错误数据。
因此我们在 CH 的查询入口层（CHProxy）添加了 QueryCache 机制，通过缓存最终查询结果以缩短查询链路，并保证了查询结果的全局可见性；同时在数据导入过程中引入版本机制，版本变更时触发存量 Cache 的自动失效，以保证结果缓存与底层数据的一致性。上线后各个 CH 集群查询平响最多下降 50%。
△Proxy QueryCache 建设
2.3 高基数精准 UV 查询
基于以上一系列的通用性查询性能优化后，虽然查询平响得到较好的优化，但是 P90、P99 指标改善却不明显。为此我们对 CH 集群的查询 SQL 进行挖掘与分析，发现大部分长尾 SQL 来自高基数的精准 UV 查询，SQL 查询格式如下：
该类查询的特点是底表数据量大、字段基数高，为了对结果去重，需要维护高成本的 HashSet。业界常见的解决思路包括：
（1）近似算法，如 HyperLogLog 预估基数，存在 1%左右的误差，大多数业务无法接受；
（2）在数据入库前预聚合，但由于 UV 的不可加和特性，在动态维度场景下不满足需求。
对查询耗时进行拆解分析，发现由于数据随机分布在各个分片上，需要在聚合层进行二次去重才能得到正确结果，聚合层成为计算热点。在整个查询过程中，子查询结果的网络传输和聚合层的二次去重耗时占到整个查询耗时的 90+%。
基于以上问题，为了在不影响查询结果准确性的前提下降低聚合层的计算负载，我们进行了两阶段的优化：
（1）NoMerge 查询：在数据入库时根据 cuid 预划分数据，查询时将聚合层的去重计算下推到各个 CH 分片，聚合层只需要执行轻量级的 NoMerge 计算；
（2）Projection+RoaringBitMap：使用 Projection 优化磁盘扫描数据量；使用 BitMap 优化 Set 计算，进一步减少中间状态大小。
2.3.1 NoMerge 查询
第一阶段我们基于数据预划分，引入了 NoMerge 查询的优化方案:
（1）在数据入库时，对数据进行预划分，保证相同 cuid 的数据落在同一个分片上，实现数据正交；
（2）改写 CH 原生的 COUNT(DISTINCT)实现逻辑：分片节点不再返回去重后的结果集，而是只上报子结果集的大小；
（3）聚合层只需要对子查询返回的结果集大小求和，即可得到最终的 UV 值。
通过这种方法，在高基数去重计数的查询场景下，可以获得 5-10 倍的性能收益。
△NoMerge 优化 COUNT(DISTINCT)
2.3.2 Projection+RoaringBitMap
第二阶段我们尝试用 Projection 结合 RoaringBitMap 进一步优化 UV 查询的性能。传统 UV 计算通过 HashSet 保存中间状态，而在高基数场景下，由于数据压缩比效果较差，预计算带来的收益不明显，因此我们引入 RoaringBitmap 数据结构替换 Hashset，减少中间状态大小，提升数据的压缩与传输效果。
△RBM 优化 COUNT(DISTINCT)
2.4 RBO 优化
参考对 UV 查询的性能优化，我们继续推进对特定类型查询的优化，以降低对应类型复杂查询的耗时。由于下游 BI 平台对常用的查询 SQL 按模板进行封装，并提供给用户直接拖拽使用，因此大量查询 SQL 具备相同的范式。对这些相同类型的查询 SQL 进行性能分析并采取针对性优化，可以以少量成本换取较大的查询性能收益，其中 RBO (基于规则的优化器) 是一种比较合适的优化选择，下面以 Case-When 查询优化为例进行说明。
Case-When 查询具体包含如下几类形式：
完整的 Case-When 查询示例如下：
该 Case-When 查询可能存在如下问题：1）无法命中索引；2）存在冗余计算。我们对此使用 RBO 规则对 Case-When 查询进行改写优化，AST 改写过程可以参考下图。上线后命中规则的 SQL 查询耗时降低 20% - 40%。
△Case-When 查询流程改写
03 ClickHouse 数据导入优化
除查询性能外，数据导入也是衡量数仓引擎性能以及易用性的关键标准之一，高效快捷的数据导入可以很大程度提升业务的数据开发与分析效率。
MEG 数据中台的导入场景以周期性离线导入为主，大部分数据为天级例行导入。我们前期主要使用原生 CH 的 Min-Batch 数据 Insert 实现数据导入，但是随着业务的覆盖以及接入数据量的增大，该导入方式暴露出以下问题：
（1）入口节点热点：类似于原生查询，原生导入首先会在集群内选择一个入口节点写数，然后入口节点根据策略将数据分发到其他分片，导入速度可能受限于入口节点的带宽；
（2）写放大：CH 底层数据采用 LSM-Tree 组织，原生导入首先在 CH 本地内存组织数据，积累到一定量后以 Part 文件形式落盘，CH 会定期对磁盘上的多个 Part 进行合并，Part 合并会带来额外的写开销。
△原生数据写入
基于以上描述可以看出 CH Min-Batch Insert 在离线导入的背景下对业务的意义不大，反而额外的数据合并会抢占一定的机器资源，影响业务查询的稳定性。因此我们建设了一套 BulkLoad 的导入机制，尽量保证"读写分离"，避免读写间的资源抢占。
BulkLoad 导入整体分为两个阶段：
（1）数据构建：离线任务从数据源摄取数据，构建为 CH 内部数据结构，而后将多个 Part 合并为一个 Part，最后将数据打包推送到分布式文件存储（后称 AFS）；
（2）数据配送：通过两阶段提交的方式对构建好的数据进行校验，并按路由规则配送到 CH 对应节点。
△数据 BulkLoad 写入
BulkLoad 导入在大规模数据导入的场景下表现出良好的吞吐，百亿行级数据导入耗时小于两小时。此外采用预计算的方式剥离数据导入需要的资源，可以更好保证稳定的查询性能。最后，我们将构建得到的数据同时作为 CH 的冷备数据，提升整体容灾能力。
此外，针对增长、反作弊等数据实时性要求较高的业务场景，我们还提供了不重不丢、秒级延迟的流式导入，目前线上单流最高导入并发超过百万 QPS。同时针对部分小数据量的临时分析场景，为复用 CH 的查询能力，我们打通了 CH 与 AFS 数据的即时查询通路。
04 ClickHouse 分布式架构升级
4.1 云原生化
MEG 数据中台前期主要基于物理机搭建 CH 集群，随着接入业务的增多以及数据规模的持续增加，集群扩容与运维效率明显满足不了规模的增长速度，具体体现如下：
（1）弹性化部署效率低：需要人工维护 CH 集群拓扑，比如人工配置集群拓扑以满足分片反亲和；
（2）大规模部署运维成本高：物理机需要自运维，并且集群扩缩容需要人工同步业务表结构。
为提升 CH 集群部署效率以及降低集群运维成本，我们将 CH 集群进行云原生化改造。与一般的无状态服务（如 Web Server 等）不同 CH 作为有状态数据库服务，云原生化改造需要解决以下问题：
（1）为追求 CH 极致性能，CH 数据存放在本地磁盘，需要保证容器重建后本地数据不丢；
（2）公司内部 PaaS 平台不支持 Kubernetes（后称 K8s）Service，需要使用其他服务发现方式， 并维护集群拓扑关系；
（3）保证分片可用性和反亲和部署。
Operator in K8s 是一个可以简化应用配置、管理和监控的 K8s 扩展方法，CH 也发起开源项目 ClickHouse-Operator，用于在 K8s 上部署和管理 CH。结合公司内现有的 PaaS 平台：Pandora、Opera、EKS 等，综合考虑接入成本、服务完整性、成熟度等因素，最终选择基于 ClickHouse-Operator 实现 CH 集群部署在 EKS。
主要实施路径为：
首先在 EKS 集群中声明 CH 的 CRD（CustomResourceDefinition）资源，然后将 Operator 部署到 EKS 集群中。当有 CH 集群创建、扩/缩容、修改 CH 配置需求时，通过 Operator 调和能力（reconcile），完成有状态的工作负载（StatefulSet）、配置管理组件（ConfigMap）和 命名服务（BNS）等资源的创建、更新。
△ClickHouse 云原生架构
针对容器重建后本地数据丢失的问题，我们采用两套数据自动恢复机制：从正常副本拷贝数据，或从 AFS 备份数据重建本地数据。业务可视自身情况权衡成本和服务稳定性，选择适合业务场景的数据透明恢复方案。
对于集群的服务发现功能，我们引入厂内经过高并发验证的 BNS 以提供相应能力。对于集群拓扑关系的维护，首先通过 StatefulSet 创建出来的 Pod 自带固定编号，以此就有了对应的分片、副本顺序；然后通过 ConfigMap 维护集群的拓扑关系，集群变更时通过自定义 Controller 监听 Pod 状态，并及时同步最新的拓扑关系。
△ClickHouse 集群拓扑关系
最后我们通过 Pod Availability 机制，结合 Readiness Gates 保障集群可用性，并通过给 Node 打标签的方式实现 Shard 反亲和部署。
目前我们已实现全量集群虚拟化部署，总规模 3000+节点，在资源成本、部署周期、维护成本都有明显的收益。
4.2 集群协调服务
虽然 CH 云原生化极大降低了集群的部署与运维成本，使得 CH 集群的规模增长速度能够满足业务的接入需求，但是规模的持续增长也对 CH 的架构稳定性和数据质量保证提出了更高的要求。在原生 CH 集群的一致性方案中，副本间日志和数据的同步主要通过 ZooKeeper（后称 ZK） 实现；分片间元数据的一致性则通过串行化执行 DDL 语句来实现。原生一致性方案在 CH 规模增长过程中暴露出以下问题：
（1）架构稳定性：基于 ZK 实现的主从架构稳定性差，ZK 容易成为集群热点；
（2）数据正确性：缺乏集群层面的事务支持，脏读、脏写导致的错误数据不如没有数据。
为解决以上问题，我们对原生 CH 架构进行了改造，其核心思路为：
（1）构建轻量级集群元信息服务（Meta），将依赖 ZK 的主从架构切换为以 Meta 为底座的无主架构，提升 CH 集群稳定性的同时消除 ZK 运维压力；
（2）基于全局 Meta 服务，结合 Quorum 协议、MVCC 机制重构数据导入、恢复、DDL 等流程，同时实现原子导入等高级功能，解决数据正确性问题。
首先是集群协调者服务的升级。集群 Meta 服务将元数据的存储和访问服务进行分层解耦，其中无状态的服务层可根据集群负载按需扩缩容，存储层保存 CH 集群的库表以及数据文件等元信息，保证 Meta 服务的全局一致性。
△ClickHouse 协调服务升级
然后是数据同步机制的升级。ZK 主要保证 CH 副本间的数据一致性，无法提供集群分片间的全局事务性，失败的数据导入任务引入的不完整数据可能导致查询时出现脏读以及数据恢复时出现脏写。为此我们引入全局的轻量级版本机制，查询、写入、数据恢复等流程根据特定版本读写数据，以提供全流程数据服务的强一致性。
△ClickHouse 数据同步机制升级
05 总结与展望
TDE-ClickHouse 作为图灵 3.0 生态中重要的基础引擎之一，凭借其强大的计算性能，为业务提供海量数据下的自助秒级分析能力。其中高性能的数据查询能力与高效的数据导入通路，可以支持下游业务更及时、敏捷地对海量数据进行分析，提供数据驱动的洞察能力，辅助业务决策；稳定可靠的分布式架构在减少资源成本与运维成本的同时，严格保障了 CH 侧的数据质量，为下游业务提供准确无误的数据反馈。
目前百度 MEG 数据中台的 CH 资源规模已超过 35 万核计算、10PB 存储；覆盖 30+业务线，单日 BI 分析查询 PV 超过 15 万，查询平响小于 3 秒，P90 小于 7 秒。
此外我们也在不断完善 MEG 数据中台 CH 的各项能力，包括存算分离减少资源与运维成本，提升规模弹性；湖仓融合降低数据迁移与探索成本；以及特定场景与类型的查询性能优化等。
随着 AI 等领域的发展，数据规模定会进一步激增，数据的重要性也会愈发明显，未来数据使用需求的变更、新硬件的出现、以及数据库相关技术理论的发展也会不断推动数据产品形态的更迭，我们也会持续关注与跟进。
————END————
推荐阅读
HelixFold 3 全球首个完整复现 AlphaFold 3，百度智能云 CHPC 为人类生命探索提供算力平台支撑
百度官方技术账号 2021-01-22 加入
关注我们，带你了解更多百度技术干货。
评论