大厂实时数仓建设项目实例

•  技术方案

• 实时计算引擎

• 实时存储引擎

• App 层

 ① 分布式-高可用

我们这里听取的是 Clickhouse 官方的建议，借助 ZK 实现高可用的方案。数据写入一个分片，仅写入一个副本，然后再写 ZK，通过 ZK 告诉同一个分片的其他副本，其他副本再过来拉取数据，保证数据一致性。

这里没有选用消息队列进行数据同步，是因为 ZK 更加轻量级。而且写的时候，任意写一个副本，其它副本都能够通过 ZK 获得一致的数据。而且就算其它节点第一次来获取数据失败了，后面只要发现它跟 ZK 上记录的数据不一致，就会再次尝试获取数据，保证一致性。

 ②海量数据-写入

数据写入遇到的第一个问题是，海量数据直接写入 Clickhouse 的话，会导致 ZK 的 QPS 太高，解决方案是改用 Batch 方式写入。Batch 设置多大呢，Batch 太小的话缓解不了 ZK 的压力，Batch 也不能太大，不然上游内存压力太大，通过实验，最终我们选用了大小几十万的 Batch。

第二个问题是，随着数据量的增长，单 QQ 看点的视频内容每天可能写入百亿级的数据，默认方案是写一张分布式表，这就会造成单台机器出现磁盘的瓶颈，尤其是 Clickhouse 底层运用的是 Mergetree，原理类似于 HBase、RocketsDB 的底层 LSM-Tree。在合并的过程中会存在写放大的问题，加重磁盘压力。峰值每分钟几千万条数据，写完耗时几十秒，如果正在做 Merge，就会阻塞写入请求，查询也会非常慢。我们做的两个优化方案：一是对磁盘做 Raid，提升磁盘的 IO；二是在写入之前进行分表，直接分开写入到不同的分片上，磁盘压力直接变为 1/N。

第三个问题是，虽然我们写入按照分片进行了划分，但是这里引入了一个分布式系统常见的问题，就是局部的 Top 并非全局 Top 的问题。比如同一个内容 ID 的数据落在了不同的分片上，计算全局 Top100 阅读的内容 ID，有一个内容 ID 在分片 1 上是 Top100，但是在其它分片上不是 Top100，导致汇总的时候，会丢失一部分数据，影响最终结果。我们做的优化是在写入之前加上一层路由，将同一个内容 ID 的记录，全部路由到同一个分片上，解决了该问题。

介绍完写入，下一步介绍 Clickhouse 的高性能存储和查询。

 ③ 高性能-存储-查询

Clickhouse 高性能查询的一个关键点是稀疏索引。稀疏索引这个设计就很有讲究，设计得好可以加速查询，设计不好反而会影响查询效率。我根据我们的业务场景，因为我们的查询大部分都是时间和内容 ID 相关的，比如说，某个内容，过去 N 分钟在各个人群表现如何？我按照日期，分钟粒度时间和内容 ID 建立了稀疏索引。针对某个内容的查询，建立稀疏索引之后，可以减少 99%的文件扫描。

还有一个问题就是，我们现在数据量太大，维度太多。拿 QQ 看点的视频内容来说，一天流水有上百亿条，有些维度有几百个类别。如果一次性把所有维度进行预聚合，数据量会指数膨胀，查询反而变慢，并且会占用大量内存空间。我们的优化，针对不同的维度，建立对应的预聚合物化视图，用空间换时间，这样可以缩短查询的时间。

分布式表查询还会有一个问题，查询单个内容 ID 的信息，分布式表会将查询下发到所有的分片上，然后再返回查询结果进行汇总。实际上，因为做过路由，一个内容 ID 只存在于一个分片上，剩下的分片都在空跑。针对这类查询，我们的优化是后台按照同样的规则先进行路由，直接查询目标分片，这样减少了 N-1/N 的负载，可以大量缩短查询时间。而且由于我们是提供的 OLAP 查询，数据满足最终一致性即可，通过主从副本读写分离，可以进一步提升性能。

我们在后台还做了一个 1 分钟的数据缓存，针对相同条件查询，后台就直接返回了。

 ④ 扩容

这里再介绍一下我们的扩容的方案，调研了业内的一些常见方案。

比如 HBase，原始数据都存放在 HDFS 上，扩容只是 Region Server 扩容，不涉及原始数据的迁移。但是 Clickhouse 的每个分片数据都是在本地，是一个比较底层存储引擎，不能像 HBase 那样方便扩容。

Redis 是哈希槽这种类似一致性哈希的方式，是比较经典分布式缓存的方案。Redis slot 在 Rehash 的过程中虽然存在短暂的 ask 读不可用，但是总体来说迁移是比较方便的，从原 h[0]迁移到 h[1]，最后再删除 h[0]。但是 Clickhouse 大部分都是 OLAP 批量查询，不是点查，而且由于列式存储，不支持删除的特性，一致性哈希的方案不是很适合。

目前扩容的方案是，另外消费一份数据，写入新 Clickhouse 集群，两个集群一起跑一段时间，因为实时数据就保存 3 天，等 3 天之后，后台服务直接访问新集群。

1) 分层设计

传统离线数仓的分层设计大家都很熟悉，为了规范的组织和管理数据，层级划分会比较多，在一些复杂逻辑处理场景还会引入临时层落地中间结果以方便下游加工处理。实时数仓考虑到时效性问题，分层设计需要尽量精简，降低中间流程出错的可能性，不过总体而言，实时数仓还是会参考离线数仓的分层思想来设计。

实时数仓分层架构如下图所示 ：

① ODS（实时数据接入层）

ODS 层，即实时数据接入层，通过数据采集工具收集各个业务系统的实时数据，对非结构化的数据进行结构化处理，保存原始数据，几乎不过滤数据；该层数据的主要来源有三个部分：第一部分是业务方创建的 NSQ 消息，第二部分是业务数据库的 Binlog 日志，第三部分是埋点日志和应用程序日志，以上三部分的实时数据最终统一写入 Kafka 存储介质中。

ODS 层表命名规范：部门名称.应用名称.数仓层级主题域前缀数据库名/消息名

例如：接入业务库的 Binlog

实时数仓表命名：deptname.appname.ods_subjectname_tablename

例如：接入业务方的 NSQ 消息

实时数仓表命名：deptname.appname.ods_subjectname_msgname

② DWS（实时明细中间层）

DWS 层，即实时明细中间层，该层以业务过程作为建模驱动，基于每个具体的业务过程事件来构建最细粒度的明细层事实表；比如交易过程，有下单事件、支付事件、发货事件等，我们会基于这些独立的事件来进行明细层的构建。在这层，事实明细数据同样是按照离线数仓的主题域来进行划分，也会采用维度建模的方式组织数据，对于一些重要的维度字段，会做适当冗余。基于有赞实时需求的场景，重点建设交易、营销、客户、店铺、商品等主题域的数据。该层的数据来源于 ODS 层，通过 FlinkSQL 进行 ETL 处理，主要工作有规范命名、数据清洗、维度补全、多流关联，最终统一写入 Kafka 存储介质中。

DWS 层表命名规范：部门名称.应用名称.数仓层级_主题域前缀_数仓表命名

例如：实时事件 A 的中间层

实时数仓表命名：

deptname.appname.dws_subjectname_tablename_eventnameA

例如：实时事件 B 的中间层

实时数仓表命名：

deptname.appname.dws_subjectname_tablename_eventnameB

③ DIM（实时维表层）

DIM 层，即实时维表层，用来存放维度数据，主要用于实时明细中间层宽化处理时补全维度使用，目前该层的数据主要存储于 HBase 中，后续会基于 QPS 和数据量大小提供更多合适类型的存储介质。

DIM 层表命名规范：应用名称_数仓层级_主题域前缀_数仓表命名

例如：HBase 存储，实时维度表

实时数仓表命名：appname_dim_tablename

④ DWA（实时汇总层）

DWA 层，即实时汇总层，该层通过 DWS 层数据进行多维汇总，提供给下游业务方使用，在实际应用过程中，不同业务方使用维度汇总的方式不太一样，根据不同的需求采用不同的技术方案去实现。第一种方式，采用 FlinkSQL 进行实时汇总，将结果指标存入 HBase、MySQL 等数据库，该种方式是我们早期采用的方案，优点是实现业务逻辑比较灵活，缺点是聚合粒度固化，不易扩展；第二种方式，采用实时 OLAP 工具进行汇总，该种方式是我们目前常用的方案，优点是聚合粒度易扩展，缺点是业务逻辑需要在中间层预处理。

DWA 层表命名规范：应用名称_数仓层级_主题域前缀_聚合粒度_数据范围

例如：HBase 存储，某域当日某粒度实时汇总表

实时数仓表命名：appname_dwa_subjectname_aggname_daily

⑤ APP（实时应用层）

APP 层，即实时应用层，该层数据已经写入应用系统的存储中，例如写入 Druid 作为 BI 看板的实时数据集；写入 HBase、MySQL 用于提供统一数据服务接口；写入 ClickHouse 用于提供实时 OLAP 服务。因为该层非常贴近业务，在命名规范上实时数仓不做统一要求。

2) 实时 ETL

实时数仓 ETL 处理过程所涉及的组件比较多，接下来盘点构建实时数仓所需要的组件以及每个组件的应用场景。如下图所示：

具体实时 ETL 处理流程如下图所示：

 ① 维度补全

创建调用 Duboo 接口的 UDF 函数在实时流里补全维度是最便捷的使用方式，但如果请求量过大，对 Duboo 接口压力会过大。在实际应用场景补全维度首选还是关联维度表，但关联也存在一定概率的丢失问题，为了弥补这种丢失，可以采用 Duboo 接口调用兜底的方式来补全。伪代码如下：

create function call_dubbo as 'XXXXXXX';create function get_json_object as 'XXXXXXX';

case    when cast( b.column as bigint) is not null        then cast( b.column as bigint)            else cast(coalesce(cast(get_json_object(call_dubbo('clusterUrl'                                                              ,'serviceName'                                                              ,'methodName'                                                              ,cast(concat('[',cast(a.column as varchar),']') as varchar)                                                              ,'key'                                                              )                                             ,'rootId')                                         as bigint)                                   ,a.column)                            as bigint)  end

 ② 幂等处理

实时任务在运行过程中难免会遇到执行异常的情况，当任务异常重启的时候会导致部分消息重新发送和消费，从而引发下游实时统计数据不准确，为了有效避免这种情况，可以选择对实时消息流做幂等处理，当消费完一条消息，将这条消息的 Key 存入 KV，如果任务异常重启导致消息重新发送的时候，先从 KV 判断该消息是否已被消费，如果已消费就不再往下发送。伪代码如下：

create function idempotenc as 'XXXXXXX';

insert into tableselect    order_nofrom    (        select            a.orderNo                                        as  order_no          , idempotenc('XXXXXXX', coalesce( order_no, '') )  as  rid        from            table1    ) twhere    t.rid = 0;

 ③ 数据验证

由于实时数仓的数据是无边界的流，相比于离线数仓固定不变的数据更难验收。基于不同的场景，我们提供了 2 种验证方式，分别是：抽样验证与全量验证。如图 3.3 所示

该方案主要应用在数据准确性验证上，实时汇总结果是基于存储在 Kafka 的实时明细中间层计算而来，但 Kafka 本身不支持按照特定条件检索，不支持写查询语句，再加上消息的无边界性，统计结果是在不断变化的，很难寻找参照物进行比对。鉴于此，我们采用了持久化消息的方法，将消息落盘到 TiDB 存储，基于 TiDB 的能力对落盘的消息进行检索、查询、汇总。编写固定时间边界的测试用例与相同时间边界的业务库数据或者离线数仓数据进行比对。通过以上方式，抽样核心店铺的数据进行指标准确性验证，确保测试用例全部通过。

该方案主要应用在数据完整性和一致性验证上，在实时维度表验证的场景使用最多。大体思路：将存储实时维度表的在线 HBase 集群中的数据同步到离线 HBase 集群中，再将离线 HBase 集群中的数据导入到 Hive 中，在限定实时维度表的时间边界后，通过数据平台提供的数据校验功能，比对实时维度表与离线维度表是否存在差异，最终确保两张表的数据完全一致。

 ④ 数据恢复

实时任务一旦上线就要求持续不断的提供准确、稳定的服务。区别于离线任务按天调度，如果离线任务出现 Bug，会有充足的时间去修复。如果实时任务出现 Bug，必须按照提前制定好的流程，严格按照步骤执行，否则极易出现问题。造成 Bug 的情况有非常多，比如代码 Bug、异常数据 Bug、实时集群 Bug，如下图展示了修复实时任务 Bug 并恢复数据的流程。

在数仓体系中会有各种各样的大数据组件，譬如 Hive/HBase/HDFS/S3，计算引擎如 MapReduce、Spark、Flink，根据不同的需求，用户会构建大数据存储和处理平台，数据在平台经过处理和分析，结果数据会保存到 MySQL、Elasticsearch 等支持快速查询的关系型、非关系型数据库中，接下来应用层就可以基于这些数据进行 BI 报表开发、用户画像，或基于 Presto 这种 OLAP 工具进行交互式查询等。

1) Lambda 架构的痛点

在整个过程中我们常常会用一些离线的调度系统，定期的（T+1 或者每隔几小时）去执行一些 Spark 分析任务，做一些数据的输入、输出或是 ETL 工作。离线数据处理的整个过程中必然存在数据延迟的现象，不管是数据接入还是中间的分析，数据的延迟都是比较大的，可能是小时级也有可能是天级别的。另外一些场景中我们也常常会为了一些实时性的需求去构建一个实时处理过程，比如借助 Flink+Kafka 去构建实时的流处理系统。

整体上，数仓架构中有非常多的组件，大大增加了整个架构的复杂性和运维的成本。

如下图，这是很多公司之前或者现在正在采用的 Lambda 架构，Lambda 架构将数仓分为离线层和实时层，相应的就有批处理和流处理两个相互独立的数据处理流程，同一份数据会被处理两次以上，同一套业务逻辑代码需要适配性的开发两次。Lambda 架构大家应该已经非常熟悉了，下面我就着重介绍一下我们采用 Lambda 架构在数仓建设过程中遇到的一些痛点问题。

例如在实时计算一些用户相关指标的实时场景下，我们想看到当前 pv、uv 时，我们会将这些数据放到实时层去做一些计算，这些指标的值就会实时呈现出来，但同时想了解用户的一个增长趋势，需要把过去一天的数据计算出来。这样就需要通过批处理的调度任务来实现，比如凌晨两三点的时候在调度系统上起一个 Spark 调度任务把当天所有的数据重新跑一遍。

很显然在这个过程中，由于两个过程运行的时间是不一样的，跑的数据却相同，因此可能造成数据的不一致。因为某一条或几条数据的更新，需要重新跑一遍整个离线分析的链路，数据更新成本很大，同时需要维护离线和实时分析两套计算平台，整个上下两层的开发流程和运维成本其实都是非常高的。

为了解决 Lambda 架构带来的各种问题，就诞生了 Kappa 架构，这个架构大家应该也非常的熟悉。

2) Kappa 架构的痛点

我们来讲一下 Kappa 架构，如下图，它中间其实用的是消息队列，通过用 Flink 将整个链路串联起来。Kappa 架构解决了 Lambda 架构中离线处理层和实时处理层之间由于引擎不一样，导致的运维成本和开发成本高昂的问题，但 Kappa 架构也有其痛点。

首先，在构建实时业务场景时，会用到 Kappa 去构建一个近实时的场景，但如果想对数仓中间层例如 ODS 层做一些简单的 OLAP 分析或者进一步的数据处理时，如将数据写到 DWD 层的 Kafka，则需要另外接入 Flink。同时，当需要从 DWD 层的 Kafka 把数据再导入到 Clickhouse，Elasticsearch，MySQL 或者是 Hive 里面做进一步的分析时，显然就增加了整个架构的复杂性。

其次，Kappa 架构是强烈依赖消息队列的，我们知道消息队列本身在整个链路上数据计算的准确性是严格依赖它上游数据的顺序，消息队列接的越多，发生乱序的可能性就越大。ODS 层数据一般是绝对准确的，把 ODS 层的数据发送到下一个 kafka 的时候就有可能发生乱序，DWD 层再发到 DWS 的时候可能又乱序了，这样数据不一致性就会变得很严重。

第三，Kafka 由于它是一个顺序存储的系统，顺序存储系统是没有办法直接在其上面利用 OLAP 分析的一些优化策略，例如谓词下推这类的优化策略，在顺序存储的 Kafka 上来实现是比较困难的事情。

那么有没有这样一个架构，既能够满足实时性的需求，又能够满足离线计算的要求，而且还能够减轻运维开发的成本，解决通过消息队列构建 Kappa 架构过程中遇到的一些痛点？答案是肯定的，后面的篇幅会详细论述。

3) 痛点总结

4) Flink+Iceberg 构建实时数仓

 ① 近实时的数据接入

前面介绍了 Iceberg 既支持读写分离，又支持并发读、增量读、小文件合并，还可以支持秒级到分钟级的延迟，基于这些优势我们尝试采用 Iceberg 这些功能来构建基于 Flink 的实时全链路批流一体化的实时数仓架构。

如下图所示，Iceberg 每次的 commit 操作，都是对数据的可见性的改变，比如说让数据从不可见变成可见，在这个过程中，就可以实现近实时的数据记录。

 ② 实时数仓 - 数据湖分析系统

此前需要先进行数据接入，比如用 Spark 的离线调度任务去跑一些数据，拉取，抽取最后再写入到 Hive 表里面，这个过程的延时比较大。有了 Iceberg 的表结构，可以中间使用 Flink，或者 spark streaming，完成近实时的数据接入。

基于以上功能，我们再来回顾一下前面讨论的 Kappa 架构，Kappa 架构的痛点上面已经描述过，Iceberg 既然能够作为一个优秀的表格式，既支持 Streaming reader，又可以支持 Streaming sink，是否可以考虑将 Kafka 替换成 Iceberg？

Iceberg 底层依赖的存储是像 HDFS 或 S3 这样的廉价存储，而且 Iceberg 是支持 parquet、orc、Avro 这样的列式存储。有列式存储的支持，就可以对 OLAP 分析进行基本的优化，在中间层直接进行计算。例如谓词下推最基本的 OLAP 优化策略，基于 Iceberg snapshot 的 Streaming reader 功能，可以把离线任务天级别到小时级别的延迟大大的降低，改造成一个近实时的数据湖分析系统。

在中间处理层，可以用 presto 进行一些简单的查询，因为 Iceberg 支持 Streaming read，所以在系统的中间层也可以直接接入 Flink，直接在中间层用 Flink 做一些批处理或者流式计算的任务，把中间结果做进一步计算后输出到下游。

总的来说，Iceberg 替换 Kafka 的优势主要包括：

优势：

• 实现存储层的流批统一；

• 中间层支持 OLAP 分析；

• 完美支持高效回溯；

• 存储成本降低。

  
  

劣势：

• 数据延迟从实时变成近实时；

• 对接其他数据系统需要额外开发工作。

秒级分析 - 数据湖加速：

由于 Iceberg 本身是将数据文件全部存储在 HDFS 上的，HDFS 读写这块对于秒级分析的场景，还是不能够完全满足我们的需求，所以接下去我们会在 Iceberg 底层支持 Alluxio 这样一个缓存，借助于缓存的能力可以实现数据湖的加速。这块的架构也在我们未来的一个规划和建设中。