【作者介绍】 王磊，阿里云 M V P ，华院计算技术总监。著有：《图解 S p a r k 大数据快速分析实战》；《 o f f e r 来了： J a v a 面试核心知识点精讲（原理篇）》；《 o f f e r 来了： J a v a 面试核心知识点精讲（架构篇）》；极客时间每日一课专栏作者。

编辑 | 韩楠

约 5 1 0 0 字 | 1 0 分钟阅读

问题背景

我们公司是一个每天百亿增量数据的互联网广告公司，作为大数据专家我主要的职责是负责系统的优化和迭代，在系统优化和迭代中，我和我们的团队一直努力寻求一种既能快速完成数据分析又能节省服务器资源的数据解决方案。

我们的广告平台的数据，包括实时数据和离线数据。实时数据首先会发送到消息系统，然后在消息系统中进行流转批，最终将转换后的数据存储在 S 3 上，离线数据则直接调用 S 3 的 A P I 将数据存储在 S 3 上。 S 3 上存储的数据格式有： C S V 、 T X T 、 P a r q u e t 、 O R C 等。

图 1

查询的时候通过 A t h e n a 、 S p a r k 或者 H a d o o p 进行查询分析。最终分析的数据用于 B I 、统计报表和告警规则等。

这种方案的好处是多源数据入库方便，并且查询的时候，在 A t h e n a 中通过标准 S Q L 按照不同的需求，可以进行任何维度的数据分析。

但是这种方案有一个很大的缺点就是查询效率慢，并且多源数据的一致性难以保障。一般对 T B 级别数据量的分析耗时在 2 0 秒以上，因此它更适合于离线数据分析。这也是大数据行业比较典型的离线数仓方案。

图 2

梳理分析，解决问题

怎么做既能够将多源数据进行融合，又能实现快速实时的分析数据。同时服务器的预算又能控制在合理的范围内，这个问题，可以说一直是我们架构优化的方向。

由于离线数据的缺点随着数据量的增大越来越被放大，因此我们迫切需要一个既能快速融合和分析多源数据，同时又经济实惠的实时数仓的方案。经过内部讨论，这个方案需要满足这样的一些条件，梳理了七点。

图 3

上面 7 个关键点， 第 3 、 4 点是目前实时数仓建立的核心需求，也是难点。

值得注意的是：

任何解决方案都没有完美的，我们只能在大量的实践摸索中不断加以完善、丰富。我们的实时数仓基于 C l i c k H o u s e 的 p r o j e c t i o n 在构建主题上探索出了比较好的方案，但是基于数据变更的自动感知，目前仍然不是个比较完美的解决方案，后续的实践中还需要进一步完善。行业里该方案其实也处于探索阶段。

言归正传，刚刚经过前面的需求分析，我们十分明确的一点是 “ 需要一个实时数仓平台 ” 。但是如何根据我们目前的需求在投入最少的情况下，获得最多的收入呢？为了达到上面的目标，我们对市场上的实时数仓方案进行了分析，最终确定基于 S t a r t R o c k s 或者 C l i c k H o u s e 构建实时数仓。来一起看看具体的实现方式：

将数据源上的实时数据直接写入消费服务。
对于数据源为离线文件的情况，有两种处理方式，一种是将文件转为流式数据写入 K a f k a ，另外一种情况是直接将文件通过 S Q L 导入 C l i c k H o u s e 集群。
C l i c k H o u s e 接入 K a f k a 消息，并将数据写入对应的原始表，基于原始表可以构建物化视图、 P r o j e c t 等实现数据聚合和统计分析。
应用服务基于 C l i c k H o u s e 数据对外提供 B I 、统计报表、告警规则等服务。

图 4

解决问题的核心关键点

在 S t a r R o c k s 、 C l i c k H o u s e 这两种方案中，我们经过性能和功能上的比较，最终选择了 C l i c k H o u s e 。这里可能有人会想了，那为什么选择 C l i c k H o u s e 呢？主要原因有这几点。

通过具体的测试发现 C l i c k H o u s e 在 5 0 T B 约 1 4 0 亿数据上，整体的查询稳定性和性能优于 S t a r R o c k s 。
C l i c k H o u s e 支持源数据载入，具体包括 K a f k a 、 S 3 、 H T T P 、 J D B C 等。
C l i c k H o u s e 支持联邦查询，可以轻松将 M y S Q L 或者 M o n g o D B 的数据和 C l i c k H o u s e 的数据进行关联查询。
基于 p r o j e c t 支持数据的实时统计。
基于物化视图支持基于原始的统计数据存储。
支持数据权限。

图 5

介绍完了我们选择 C l i c k H o u s e 的原因，下面进入到本文的重头戏，如何通过 C l i c k H o u s e 实现我们最初在构建实时数仓时提出的要求。

（一）如何保障各种来源的数据能实时地入仓。

一般我们将多源数据分为离线数据、实时数据和其他周边库数据。离线数据主要指文件数据，实时数据主要是 K a f k a 消息队列中的数据、周边库数据指的的类似 M y S Q L 、 M o n g o D B 等数据。接下来从实战角度，分别说下如何接入各种数据源。

（ 1 ） M y S Q L 的数据接入，可以通过使用 J D B C 方式实现。

CREATE TABLE user_table(    `id` Int32,    `user_name` String,    `height` Float32,    `password` Nullable(String))ENGINE JDBC('jdbc:mysql://localhost:3306/?user=root&password=root', 'test', 'test')

c o d e 1

通过前面代码将 M y S Q L 表接入到 C l i c k H o u s e 后，就可以直接在 C l i c k H o u s e 中直接查询 M y S Q L 表中的数据。

SELECT * FROM user_table

c o d e 2

同样也可以向 M y S Q L 中插入数据。

INSERT INTO user_table(`id`, `user_name`) VALUES(1,'alex')

c o d e 3

（ 2 ）下面咱们看下 M o n g o D B 的数据接入：

CREATE TABLE [IF NOT EXISTS] testdb.test_collection(    id UInt64,    name String,) ENGINE = MongoDB(127.0.0.1:27017, testdb, test_collection, 'your_user_name', 'your_password');

c o d e 4

表建立好后便可以执行查询语句。

SELECT COUNT() FROM test_collection;

c o d e 5

（ 3 ）再看下 S 3 的数据接入：

CREATE TABLE s3_table (name String, value UInt32)     ENGINE=S3(')    SETTINGS input_format_with_names_use_header = 0;

c o d e 6

S 3 表数据建立好后，我们向其中插入数据。

INSERT INTO s3_table VALUES ('a', 1), ('b', 2), ('c', 3);

c o d e 7

同样可以像使用关系型数据库一样查询 S 3 数据表。

SELECT * FROM s3_table LIMIT 2;

c o d e 8

（ 4 ) K a f k a 数据接入： C l i c k H o u s e 提供了 K a f k a 数据接入引擎，可以方便地实现将 K a f k a 的数据接入到 C l i c k H o u s e 中，并将其插入到 C l i c k H o u s e 表中。

CREATE TABLE kafka_source_test (    level String,    type String,   name String,   time DateTime64  ) ENGINE = Kafka SETTINGS kafka_broker_list = 'localhost:9092',                            kafka_topic_list = 'test_topic',                            kafka_group_name = 'test_group',                            kafka_format = 'JSONEachRow',                            kafka_num_consumers = 4;

c o d e 9

这段代码接入了地址为 l o c a l h o s t : 9 0 9 2 ， t o p i c 为 t e s t _ t o p i c 的 K a f k a 实时数据。其中消费者分组为 t e s t _ g r o u p ， C l i c k H o u s e 会实时将数据消费到 C l i c k H o u s e 的表 k a f k a _ s o u r c e _ t e s t 中。当 K a f k a 的 T o p i c 有数据产生时，数据会实时被消费和处理，并插入 k a f k a _ s o u r c e _ t e s t 表中。

  SELECT * FROM kafka_source_test LIMIT 5

c o d e 1 0

通过前面的介绍，你可以看到 C l i c k H o u s e 接入数据源还是比较简单的，其中 K a f k a 是构建实时数仓最常用的数据源。

（二）数据入仓能及时被分析。

这里的数据分析分为两种情况，一种情况是有明确需求的报表分析，这种情况下可以对实时数据边入库边分析，具体可以通过建立物化视图的方式实现，也可以通过 C l i c k H o u s e 的 p r o j e c t i o n 实现。另外一种情况是随机性的数据分析需求，像这种需求 C l i c k H o u s e 基于索引可以做到 T B 级别数据秒级计算返回。具体的代码实现后面将会介绍。

（三）基于实时数仓能方便的根据业务需求建立主题数据。

我们构建实时数仓的很大一个初衷是：当客户有基于主题的报表需求时，我们通过一个配置或者一条 S Q L 语句，就能实现业务的统计报表。那么具体在 C l i c k H o u s e 中如何实现呢？

其中物化视图和 p r o j e c t i o n ，是 C l i c k H o u s e 实现灵活的按照业务构建主题数据的具体解决方案，下面咱们分别对这两个关键知识点进行介绍。

其他更多 C l i c k H o u s e 的原理和使用，需要同学们在具体使用中查阅官方文档（ h t t p s : / / c l i c k h o u s e . c o m / d o c s / e n / ）。

1 . 物化视图

虽然 C l i c k H o u s e 的性能在 T B 级别的数据的查询上，已经能达到秒级返回了，近乎完美。但是在超大规模和特别复杂的计算场景下计算耗时，还是面对一些挑战，这时我们就需要通过一个类似触发器的机制，将原始数据进行实时汇总，这样我们统计分析时直接查询汇总后的表就可以，如此一来，无论对于服务器负载还是业务快速分析的需求都是友好的。那么如何对 C l i c k H o u s e 数据进行实时分析汇总呢？答案就是物化视图。

这里得先看下物化视图的原理。

物化视图的构建过程中包括一个原表，一个物化视图和一个目标表。物化视图定义了数据物化的计算逻辑，当我们向原表插入一条数据时，会触发物化计算，计算完成后将计算结果实时写入目标表。用户查询的时候查询目标表即可完成数据的分析。

图 6

在实际应用中我们对于报表需求，我们通过物化视图来实现，例如对于 k a f k a _ s o u r c e _ t e s t 中的数据，我们需要基于 t y p e 字段按照 “ 天 ” 进行统计。

首先我们需要建立一个物化视图的基础表。

CREATE TABLE statistics_type_day(    type String,    day Date,    type_count UInt32)ENGINE = SummingMergeTree()ORDER BY (type, day)

c o d e 1 1

上述建表语句中使用了 S u m m i n g M e r g e T r e e 引擎，该引擎用于统计数据的存储，它能在合并分区的时候，按照预定义条件对数据进行聚合，将相同分组的多行数据统一到一行，从而显著减少存储空间，加快查询的效率。

接下来我们创建一个物化视图，将 k a f k a _ s o u r c e _ t e s t 表中的数据实时进行统计分析并写入 s t a t i s t i c s _ t y p e _ d a y 表中。

CREATE MATERIALIZED VIEW if not exists statistics_type_day_mv TO statistics_type_day AS SELECT type, day, count(1) type_countfrom kafka_source_testgroup by type, day

c o d e 1 2

这样当 k a f k a _ s o u r c e _ t e s t 中有数据的时候，便能实时统计并将统计结果物化下来。查询的时候直接查询物化表 s t a t i s t i c s _ t y p e _ d a y 就行。

select * from statistics_type_day where day='2022-03-20'

c o d e 1 3

2 . p r o j e c t i o n ：数据投影

C l i c k H o u s e 的 p r o j e c t i o n 本质上实现的是数据聚合，小伙伴可能要问了，既然 C l i c k H o u s e 已经有物化视图了，为何还需要 p r o j e c t i o n 呢？

试想这样一个场景，我们有一个 1 5 0 个字段的表，基于该表我们针对不同业务构建了 2 0 个物化视图。那我们业务在分析的时候如果查询基础数据就到基础表查询，如果查询的是统计数据则需要在对应的物化视图的统计表中查询。

但是在实际的使用过程中，常常因为构建视图和使用视图的人不在一个团队，一种情况是使用者不知道有哪些视图可以使用，所以在写 S Q L 的是时候就直接查基表了，另外一种情况是业务需要针对基础表的 S Q L 和统计表的 S Q L 维护多个业务 S Q L ，在查询上不统一。

图 7

那么如何构建一个表，在该表上既能完成基础数据的查询，也能进行统计数据的查询，如果能实现的话，之前的 1 + 2 0 个表 S Q L 的维护就变成了 1 张表 S Q L 的维护。使用起来将大大简化应用层开发，应用层也不用关注哪些是统计数据，哪些是基础数据。

C l i c k H o u s e 的 p r o j e c t i o n 给了这个问题一个很好的解决方案。

p r o j e c t i o n 在数据分析功能上类似于物化视图，我们可以在 p r o j e c t i o n 中预定义表达式，当数据写入的时候，会将原始数据和针对 p r o j e c t i o n 中表达式计算后的聚合数据一起写入存储。

在查询的时候会经过一个智能路由的过程，智能路由通过分析 S Q L 语句，如果发现 S Q L 语句查询的数据都在聚合数据中，则直接查询聚合数据并将结果返回给用户，这块便大大减少了服务器资源的开销，只有到查询的数据在各个 p r o j e c t i o n 都不存在的时候才会去查基表进行统计。也可以理解为 C l i c k H o u s e 基于 p r o j e c t i o n 自动完成了查询优化。（具体参见下图）

图 8

了解了 p r o j e c t i o n 原理后，接下来我们看下如何使用 p r o j e c t i o n ，首先需要定义 p r o j e c t i o n 。

CREATE TABLE test_projection_table(    level String,   type String,   name String,   city String,   time DateTime64,    PROJECTION projection_1    (         SELECT             name,             count(1)         GROUP BY  level    ) ,     PROJECTION projection_2    (         SELECT             name,             count(1)         GROUP BY type     ) ) ENGINE = MergeTree() ORDER BY (name , level, type)

c o d e 1 4

上述代码在定义表的时候定义了 p r o j e c t i o n _ 1 和 p r o j e c t i o n _ 2 ，其中 p r o j e c t i o n _ 1 在 l e v e l 维度对数据进行汇总， p r o j e c t i o n _ 2 在 t y p e 维度对数据进行汇总。在创建好 p r o j e c t i o n 后还可以对 p r o j e c t i o n 进行修改。

ALTER TABLE test_projection_table     ADD PROJECTION projection_3    (         SELECT             count(1),             name        GROUP BY type, level     )

c o d e 1 5

那么查询的时候，如何才能命中 p r o j e c t i o n 呢？得满足这样几个条件：

s e l e c t 表达式必须为 p r o j e c t i o n 定义中 s e l e c t 表达式的子集。
g r o u p b y 列必须为 p r o j e c t i o n 定义中 g r o u p b y 列的子集。
w h e r e 条件必须为 p r o j e c i t o n 定义中的 g r o u p b y 列的子集。

例如这里 S Q L 查询的时候会走表达式：

select name,count() from test_projection_table where type='A' group by type

c o d e 1 6

而 S Q L 不会走 p r o j e c t i o n ，因为 c i t y 不在 p r o j e c t i o n 的定义中。

select name,city,count() from test_projection_table where type='A' group by city

c o d e 1 7

具体可以通过 e x p l a i n 查看执行计划，如果出现 R e a d F r o m S t o r a g e ( M e r g e T r e e ( w i t h p r o j e c t i o n ) ) ，表示命中 p r o j e c t i o n 。

（四）实时数仓需要提供元数据变更的自动感知能力。

对于元数据的变更 C l i c k H o u s e 无法做到自动感知，我们构建了一个元数据管理平台，实时监 C l i c k H o u s e I n s e r t 语句，当发现有新的 S c h e m a 变更后以告警的形式通知研发人员，有研发人员修改表结构以兼容元数据的变更。

（五）实时数仓难免需要要外部数据交互，因此实时数仓需要具备联邦查询的能力。

对于联邦查询，我们将外部库表数据接入 C l i c k H o u s e 并和 C l i c k H o u s e 进行联邦查询。

（六）对外提供一个统一的数据分析服务，用户不用关心数仓底层数据复杂的关系关系。

在查询上我们利用 p r o j e c t i o n 特性将查询统一到一个表上，除非有特别特殊的需求会走物化视图。

（七）需要确保实时数仓中存储的数据量在合理的访问内，以确保不会产生过多的服务器资源账单。

在控制数据的存储量上我们通过 C l i c k H o u s e 的冷热分离思想，将热数据存储在 C l i c k H o u s e 上，将冷数据存储在 S 3 上，以控制服务器资源。

图 9

结束语

本文我介绍了如何基于 C l i c k H o u s e 的百亿级广告平台实时数仓构建实战。在数仓的构建前期我们首先基于数据要达成的目标进行了梳理。

图 1 0

我们在 S t a r R o c k s 、 C l i c k H o u s e 选型上，基于性能原因我们选择了 C l i c k H o u s e 。在具体的实现细节上我们没有像传统数仓那样构建完成的 O D S 、 D W D 、 D W S 、 A D S 层，而是借助 C l i c k H o u s e 的物化视图和 p r o j e c t i o n 完成我们构建实时数仓的目标。

其实在该架构中，可以将接入 K a f k a 原表和外部表的数据理解为 O D S （贴源数据层），将经过数据清洗转换后的基础表理解为 D W D （数据明细层），将基于基础表衍生的物化视图和 p r o j e c t i o n 等统计数据理解为 D W S （轻度汇总层），基于视图我们还可以接着构建视图，表和 p r o j e c t i o n 。

图 1 1

延伸思考

前面给大家介绍了我们的实时数仓的方案，那么实时数仓是如何演化来的呢？下面多说几句，跟大家从宏观上说下大数据的行业发展，以便于能够从一个更高视角理解实时数仓。只有大家了解了大数据的发展历程，才能更好地理解在当下企业建立实时数仓的必要性，以及我们未来实时数仓未来的发展方向。

快速方便地构建基于主题的统计数据，一直是大数据行业的痛点，最初解决该问题的方案是 H a d o o p 和 S p a r k 的离线计算，但随着业务的发展，这两种方案在数据实时性上的短板越来越明显。行业急需一个实时数仓方案，但是在实时数仓中如果基于 F l i n k 进行实时计算的话，对于业务的频繁变更带来的开发成本又变得不可控。因此大家又把目光投向了 M P P 架构，这种架构基本上达到了 1 个 S Q L 就能满足一个业务报表的需求，方便快速，数据又是实时计算的，基本满足了我们的要求。但是其在分布式事务和元数据的自动感知上，还有待完善。

从上面的趋势中也可以看到大数据发展的阶段，第一阶段大家基于大型 M y S Q L 的分库分表实现大规模数据的存储和分析，但是随着数据量的增大，这个方案玩不转了。于是出现了 H a d o o p 和 S p a r k 来解决海量数据计算问题。等海量数据计算的需求满足后大家又对数据的实时性要求更高了，业务总想看着最新的数据，于是出现了实时数仓。

图 1 2

但是问题到这里并没有完，后续必然在实时数仓的分布式数据一致性，事务的支持，元数据的动态管理和数据治理上，有更多的需求出现。这个也是大数据行业下个阶段要解决的重点问题。

因此大数据不像关系型数据库那样有稳定的解决方案，而是根据行业的变化和用户的需求在不断的迭代和更新。其最终目标是既能像关系数据库那样满足 A C I D 的需求，又具有大规模数据实时计算的能力和灵活的数据分析能力。

最后要说的是 C l i c k H o u s e 之所以性能优异，是因为它采用了向量化计算技术。将多次 C P U 计算优化为一次 C P U 计算，从而大大提升了 C P U 的效率。具体实现技术手段为采用 S I M D ( S i n g l e I n s t r u c t i o n M u l t i p l e D a t a ) 技术实现单条指令操作多条数据，在 C P U 寄存器层面实现数据的并行操作。

向量化计算可以说是 M P P 架构的 “ 银弹 ” ，目前 C l i c k H o u s e 和 S t a r R o c k s 都采用了这个技术方案来达到海量数据快速分析的，后续肯定会有更多的数据库方案跟进该技术。限于篇幅问题在这里不能为大家详细介绍向量化计算的魅力，后续如果大家感兴趣我再单独拿出来讲一下。

最后，欢迎你在评论区留言与交流，也可以把这一篇分享给你的朋友，我们下期再会。

基于ClickHouse的百亿级广告平台实时数仓构建实战

（ 一 ） 如 何 保 障 各 种 来 源 的 数 据 能 实 时 地 入 仓。

（ 二 ） 数 据 入 仓 能 及 时 被 分 析。

（ 三 ） 基 于 实 时 数 仓 能 方 便 的 根 据 业 务 需 求 建 立 主 题 数 据。

（ 四 ） 实 时 数 仓 需 要 提 供 元 数 据 变 更 的 自 动 感 知 能 力 。

（ 五 ） 实 时 数 仓 难 免 需 要 要 外 部 数 据 交 互 ， 因 此 实 时 数 仓 需 要 具 备 联 邦 查 询 的 能 力 。

（ 六 ） 对 外 提 供 一 个 统 一 的 数 据 分 析 服 务 ， 用 户 不 用 关 心 数 仓 底 层 数 据 复 杂 的 关 系 关 系 。