摘要：本文整理自阿里云资深技术专家，阿里云Hologres负责人姜伟华，在FFA实时湖仓专场的分享。点击查看>>视频回放：

本篇内容主要分为四个部分：

一、实时数仓分层的技术需求

二、阿里云一站式实时数仓Hologres介绍

三、Flink x Hologres：天作之合

四、基于Flink Catalog的Streaming Warehouse实践

一、实时数仓分层的技术需求

首先，我们讲一讲数仓的分层技术以及分层技术的现状。

1、实时数仓分层技术现状

大数据现在越来越讲究实时化，在各种场景下都需要实时，例如春晚实时直播大屏，双11 GMV实时大屏、实时个性化推荐等场景，都对数据的实时性有着非常高的要求。为了满足业务的实时性需求，大数据技术也开始逐步发展出实时数仓。

但如何构建实时数仓呢？

相比离线数仓，实时数仓没有明确的方法论体系。因此在实践中，有各种各样的方法，但没有一个方法是万能。最近行业内提出了Streaming Warehouse的概念。Streaming Warehouse的本质是分层之间能够做到实时数据的流动，从而解决实时数仓分层的问题。

下面，我们先来了解下实时数仓的主流分层方案。

2、实时数仓主流分层方案

实时数仓的主流分层方案主要有4个。

方案1：流式ETL

ETL（Extract- Transform-Load）是比较传统的数据仓库建设方法，而流式ETL就是指：实时数据经过Flink实时ETL处理之后，将结果写入到KV引擎中，供应用查询。而为了解决中间层不方便排查的问题，也需要将中间层数据同步到实时数仓中供分析之用。最常见的做法就是数据通过Flink清洗后，写到Kafka形成ODS层。再从Kafka消费，经过加工形成DWD层。然后Flink加工成DWS层，最后通过加工形成ADS层的数据写到KV引擎并对接上层应用。因为直接使用Kafka数据进行分析和探查很麻烦，所以也会同步一份Kafka数据到实时数仓，通过实时数仓进行分析和探查。

这个方案的优势是层次明确，分工明确。但劣势是需要有大量的同步任务、数据资源消耗很大、数据有很多冗余、处理链路较复杂需要很多的组件。除此之外，这个方案构建的实时数仓分层，尤其是Kafka分层，复用性非常差，也没办法响应schema的动态变化。

方案2：流式ELT

而流式ELT则是将计算后置，直接将明细数据写进实时数仓（EL），不需要严格的数仓分层，整个架构只需要一层，上层应用查询的时候进行数据的变换（T）或者分层。常见的做法就是把数据加工清洗后，写到实时数仓里，形成DWD层，所有的查询都基于DWD层的明细数据进行。

这个方案的好处在于，没有ETL，只有一层；数据修订很方便。但它的弊端有两个方面：

• 在查询性能方面，由于是明细数据查询，所以在某些场景下不能满足QPS或延迟的要求。
• 因为没有严格的数仓分层，所以数据复用很困难，很难兼顾各方面的诉求。

方案3：定时调度

既然实时流式无法完成数据的实时数仓分层，我们可以将数据实时写入实时数仓的DWD层。DWS层、ADS层用离线的高频调度方法，实现分钟级的调度，从而借用离线数仓，进行分层构造。这个也就是业界常用的方案3。

这个方案的好处在于可以复用很多离线经验，方案成本低且成熟。但方案也存在如下缺点：

• 延迟大：每一层的延迟都跟调度相关，随着层次越多，调度延迟越大，实时数仓也变成了准实时数仓。
• 不能完全复用离线方案：离线调度一般是小时级或天级，我们可以使用全量计算。但在分钟级调度时，必须做增量计算，否则无法及时调度。

二、阿里云一站式实时数仓Hologres介绍

接下来，先介绍一下阿里云一站式实时数仓Hologres产品。Hologres是阿里云自研的一站式实时数仓，它同时包含三种能力：

• OLAP能力：同传统的实时数仓一样，可以支持数据的实时写入、以及复杂OLAP实时多维分析快速响应，满足业务的极致数据探索能力。
• 在线服务Serving(KV)：可以支持KV查询场景，提供非常高的QPS和毫秒级的低延迟。
• 湖仓一体：能够直接查询数据湖的数据，以及能够加速阿里云离线数仓MaxCompute，助力业务更低成本实现湖仓一体。

下面为具体介绍：

• 首先，大家可以把Hologres当做一个常见的实时数仓。它的特点在于写入侧支持百万RPS的实时写入，写入即可查，没有延迟。同时也支持高性能的实时整行更新和局部更新。其中，整行更新是把整行替换掉，局部更新可以更新一行中的局部字段，二者都是实时更新。
• 在查询侧，一方面支持复杂的OLAP多维分析，可以非常好的支持实时大屏、实时报表等场景。近期Hologres拿到了TPC-H 30TB的性能世界第一的TPC官方认证成绩，见>>阿里云 ODPS-Hologres刷新世界纪录，领先第二名23%。其次，Hologres也支持在线服务查询，不仅支持百万QPS KV点查，而且也支持阿里云达摩院的Proxima向量检索引擎，可以支持非常高效的向量检索能力。同时这些能力在Hologres中是全用SQL表达，对用户使用非常友好。此外，为了兼顾数据服务和实时数仓的需求，Hologres在行存、列存的数据格式基础上，也支持行列共存，即行列共存的表即一份行存，又有一份列存，并且系统保证这两份数据是强一致的，对于OLAP分析，优化器会自动选择列存，对于线上服务，会自动选择行存，通过行列共存可以非常友好的实现一份数据支撑多个应用场景。
• 因为Hologres同时支持OLAP分析和线上服务，其中线上服务要求非常高的稳定性和SLA。为了保证OLAP分析和线上服务时不会发生冲突，我们支持了读写分离，从而实现OLAP与数据服务的强隔离。
• 最后，在湖仓数据交互式分析方面，Hologres对阿里云MaxCompute离线数仓里的数据，数据湖中的数据都可以秒级交互式分析，且不需要做任何的数据搬迁。
• 除此之外，Hologres的定位是一站式的企业级实时数仓，所以除了上述能力，我们还有很多其他能力。包括数据的治理、成本治理、数据血缘、数据脱敏、数据加密、IP白名单、数据的备份和恢复等等。

三、Flink x Hologres：天作之合

1、Hologres与Flink深度集成

Flink对于实时数仓能够提供非常丰富的数据处理、数据入湖仓的能力。Hologres与Flink有些非常深度的整合能力，具体包括：

• Hologres可以作为Flink的维表：在实时计算的场景下，Flink对维表的需求很强，Hologres支持百万级至千万级RPS的KV点查能力，可以直接当做Flink维表使用，且可以做到实时更新，对于像实时特征存储等维表关联场景就也可以非常高效的支持。
• Hologres可以作为Flink的结果表：Hologres支持高性能的实时写入和整行实时更新的能力，可以结合Flink，输出需要强大的Update能力，满足数仓场景下的实时更新、覆盖等需求。与此同时，Hologres还有很强的局部更新能力。局部更新能力在很多场景下，可以替代Flink的多流Join，为客户节省成本。
• Hologres可以作为Flink的源表：Hologres支持Binlog能力，一张表的任何变化，比如insert、update、delete等等，都会产生Binlog事件。Flink可以订阅Hologres Binlog，进行驱动计算。由于Flink支持Hologres的整表读取，二者结合构成了Flink全增量一体化的读取能力。并且，Hologres也对了接Flink CDC，它可以驱动Flink CDC的计算。
• 支持Hologres Catalog：通过Hologres Catalog的任何操作，都会直接实时反映到Hologres里，用户也不需要在Flink建Hologres表，这样就使得Flink+Hologres就具备了整库同步、Schema Evolution的能力。

2、基于Flink+Hologres的Streaming Warehouse方案

那Flink和Hologres如何构建Streaming Warehouse？

Streaming Warehouse：数据能在数仓之间实时的流动，本质上就是解决实时数仓分层的问题

最开始我们介绍了常见的数仓分层方案，Flink+Hologres的Streaming Warehouse方案则是可以完全将Flink+Kafka替换。具体做法如下：

1. 将Flink写到Hologres里，形成ODS层。Flink订阅ODS层的Hologres Binlog进行加工，将Flink从DWD层再次写入Hologres里。
2. Flink再订阅DWD层的Hologres Binlog，通过计算形成DWS层，将其再次写入Hologres里。
3. 最后，由Hologres对外提供应用查询。

该方案相比Kafka有如下优点：

• 解决了传统中间层Kafka数据不易查、不易更新、不易修正的问题。Hologres的每一层都可查、可更新、可修正。
• Hologres的每一层都可以单独对外提供服务。因为每一层的数据都是可查的，所以数据的复用会更好，真正实现数仓分层复用的目标。
• Hologres支持数据复用，模型统一，架构简化。通过Flink+Hologres，就能实现实时数仓分层，简化架构和降低成本。

3、Flink+Hologres核心能力：Binlog、行列共存、资源隔离

上面讲的Flink+Hologres的Streaming Warehouse方案，其强依赖于以下三个Hologres核心能力：

1. Binlog：因为实时数仓一般没有Binlog，但Hologres提供了Binlog能力，用来驱动Flink做实时计算，正因为有了Binlog，Hologres才能作为流式计算的上游。
2. 行列共存。一张表既有行存数据，又有列存数据。这两份数据是强一致的。行列共存的特性让中间层的每张表，不但能够给Flink使用，而且可以给其他应用（比如OLAP、或者线上服务）使用。
3. 资源强隔离。实时数仓一般是弱隔离或软隔离，通过资源组、资源队列的方法实现资源隔离。如果Flink的资源消耗很大，可能影响中间层的点查性能。但在Hologres强隔离的能力下，Flink对Hologres Binlog的数据拉取，不会影响线上服务。

通过Binlog、行列共存、资源强隔离的三个特点，不仅能让Flink+Hologres形成Streaming Warehouse，并且能够使中间的每层数据复用，被其他应用或线上服务使用，助力企业构建最简单最完整的实时数仓。

4、基于Flink+Hologres的多流合并

接下来，讲一讲基于Flink+Hologres的多流合并。

因为Hologres有特别强大的局部更新能力，基于此我们可以简化Flink的多流Join。比如在风控场景下，我们需要基于用户ID构建用户的多侧面画像，用户画像来自很多数据源，比如客户的浏览行为、成交行为、履约行为等等。把数据源的数据按照用户ID，把每个用户放到一行里，形成不同的字段，形成用户的完整画像。

传统的方式需要用Flink多流Join实现，Flink把上游的多个数据源关联到一起，Join后写到Kafka里，然后驱动下游的Flink，加工这行完整的数据。这就使得多流Join非常耗资源。

所以在Flink+Hologres的Streaming Warehouse方案中，可以利用Hologres的局部更新能力，把一张表定为定义成Hologres的行存表或行列共存表。此时，整个方案就简化成上游每个数据源，同步数据到Hologres表的若干个字段里，若干个任务同时写入这张表，然后利用Hologres的局部更新能力，把数据汇总在一起。

如果打开这张Hologres表的 Binlog，上游任何数据源的变化都会更新这张表，使这张表的Binlog中生成行数据的最新状态，然后驱动下游的Flink继续计算，从而完美匹配常见的风控场景。这种用法下，资源消耗、运维都得到了极大的简化。

四、基于Flink Catalog的Streaming Warehouse实践

Flink+Hologres的Streaming Warehouse方案已经非常成熟，但唯一的缺点在于，用户需要在两个系统之间切换，过程比较繁琐。为了让用户操作更简单，我们基于Flink Catalog提供了更加简单的使用体验。

下面我们来看看怎么样基于Flink Catalog去构建基于Flink+Hologres的Streaming Warehouse。我们会发现，有了Flink Catalog后，整个使用体验会很简单，并能充分发挥Flink和Hologres两个产品的强大能力。

下图是一个典型的Flink+Hologres实时ETL链路：

1. ODS层、DWD层、ODS层的数据都存在Hologres中。
2. 链路中所有的数据加工都是通过Flink SQL完成。在整个ETL链路中，用户不需要任何Hologres SQL，直接写Flink SQL即可。
3. Flink用户可以通过Flink SQL对每层中的Hologres数据进行数据探查（流模式和批模式都可以）。比如：当我们发现DWS层的数据结果出现问题，需要查看哪层的结果有问题或逻辑有错误。此时，我们可以复用原来的Flink SQL来进行探查、定位或者数据重新消费。
4. Hologres中的每层数据都可以对外提供查询和服务（通过Hologres SQL）。

接下来，以某个电商场景为例，演示一下基于Flink Catalog的Streaming Warehouse。如下图所示，有一个MySQL数据库作为订单库，里面有订单表orders、订单支付表orders_pay、以及产品品类表product_catalog。

1. 第一步，我们通过Flink的实时数仓，把数据实时同步到Hologres里，形成ODS层。
2. 第二步，加工DWD层。将DWD层的数据写到Hologres里。在这个过程中，我们需要把订单表和订单支付表，合并成一张表，实现多路合并。与此同时，我们希望orders表关联商品品类表product_catalog。
3. 第三步，驱动下游计算，构建DWS层。以用户维度和商店维度，收集统计数据。比如用户每天的订单金额和商店每天的订单金额，从而形成一条完整的链路。
4. 第四步，将DWS层的表推荐给系统使用。作为用户和商店的特征，用做推荐用途。
5. 第五步，DWD层的表能够直接用来做实时统计分析、统计产品、实时大屏、实时报表。

上图中的绿色链路，全部使用Flink SQL完成。橙色链路对外提供服务，由Hologres SQL完成。

接下来，讲一讲每个步骤是如何运行的。

第一步，在Flink实时数仓，形成ODS层。首先，创建一个Hologres的Catalog。MySQL中存储订单、支付以及商品信息3张表，通过Flink Catalog功能，将MySQL整库的数据实时同步至Hologres，形成ODS。相关代码如下所示。我们可以看到，MySQL整库同步到Hologres，通过Flink SQL来表达是非常简单的。

--Flink SQL-- 创建Hologres CatalogCREATE CATALOG holo WITH ( ‘type’ = ‘hologres’ … );-- MySQL整库同步到HologresCREATE DATABASE IF NOT EXISTS holo.order_dwAS DATABASE mysql.sw INCLUDING all tables;

第二步，DWD实时构建。数据实时写入ODS层后，Flink读取Hologres Binlog，并用多流合并、维表关联将订单、交易、商品3个表打成一个大宽表，实时写入至Hologres的订单汇总表中，形成DWD层。

如下SQL是DWD层表的建表语句。这张目标表包含了来自orders、orders_pay、product_catalog的字段，关联了相关的用户信息、商户信息、订单信息、商品品类信息等等，形成了一张宽表。

--Hologres SQLCREATETABLE holo.order_dw.dwd_orders(  order_id bigintnotnull primary key,--字段来自order 表  order_user_id bigint,  order_shop_id bigint,  order_product_id string,  order_fee numeric(20,2),  order_create_time timestamp_ltz,  order_update_time timestamp_ltz,  order_state int,--字段来自product_catalog表  order_product_catalog_name string,--字段来自orders_pay表  pay_id bigint,  pay_platfrom int,  pay_create_time timestamp_ltz
);

下面的SQL是真正的计算逻辑，这里包含两个INSERT语句：

• 第一个INSERT语句是从orders表实时打宽后写入。这里用到了Hologres的维表关联能力。实时打宽后，写入目标表的部分字段。
• 第二个INSERT语句是从orders_pay表实时同步到同一张目标表，更新另外一些字段。

这两个INSERT语句最大的关联在于，它们写的是同一张表，会自动利用目标表的主键ID进行关联。每个INSERT都是做了目标表的局部更新，两者的合力结果是实时更新的目标宽表。

--Flink SQLBEGIN STATEMENT SET;-- 从orders表实时打宽后写入INSERTINTO holo.order_dw.dwd_orders(order_id,order_user_id,order_shop_id,order_product_id,order_fee,order_create_time,order_update_time,order_state,order_product_catalog_name
)SELECTo.*,dim.catalog_nameFROMholo.order_dw.orders o 
LEFT JOIN holo.order_dw.product_catalogFOR SYSTEM_TIME AS OF proctime ()AS dim
ONo.product_id= dim.product_id;-- 从order_pays表实时写入INSERTINTO holo.order_dw.dwd_orders(pay_id,order_id,pay_platform,pay_create_time
)SELECT*FROMholo.order_dw.orders_pay;END;

第三步，DWS层的实时聚合。在DWD的基础上，通过Flink读取Hologres DWD的Binlog数据，进行实时指标聚合计算，比如按照用户维度聚合，按照商户维度聚合等，然后实时写入Hologres，形成DWS层。

• 先是创建对应的聚合指标表，DDL语句如下

--Flink SQL-- 用户维度聚合指标表CREATETABLE holo.order_dw.dws_users(  user_id bigintnotnull,  ds string notnull,-- 当日完成支付总金额    payed_buy_fee_sum numeric(20,2) not null,        primary key(user_id,ds)NOT ENFORCED
);-- 商户维度聚合指标表CREATETABLE holo.order_dw.dws_shops(  shop_id bigintnotnull,  ds string notnull,-- 当日完成支付总金额  payed_buy_fee_sum numeric(20,2)notnull,  primary key(shop_id,ds)NOT ENFORCED
);

• 然后将数据写入Hologres中，经过简单的三步后，Flink SQL构建了完整的Streaming Warehouse分层体系。

--flink sql--数据写入HologresBEGIN STATEMENT SET;INSERTINTO holo.order_dw.dws_usersSELECT  order_user_id,  DATE_FORMAT (pay_create_time,'yyyyMMdd')as ds,  SUM (order_fee)FROM holo.order_dw.dwd_orders c
WHERE pay_id ISNOTNULLAND order_fee ISNOTNULLGROUPBY order_user_id, DATE_FORMAT (pay_create_time,'yyyyMMdd');INSERTINTO holo.order_dw.dws_shopsSELECT  order_shop_id,  DATE_FORMAT (pay_create_time,'yyyyMMdd')as ds,  SUM (order_fee)FROM holo.order_dw.dwd_orders c
WHERE pay_id ISNOTNULLAND order_fee ISNOTNULLGROUPBY order_shop_id, DATE_FORMAT (pay_create_time,'yyyyMMdd');END;

第四步，构建应用，基于DWS层，对外提供服务。

数据的分层和加工完成后，业务就可以通过Hologres查询数据并应用。在这个例子里，推荐系统要求非常高的点查性能，所以要求百万级的QPS检查能力。Hologres的行存表或者行列共存表完全可以满足。

这个方案和传统的实时数仓最大的差别是：传统的实时数仓只有最后一层的数据，可对外提供服务。而在Hologres里，DWD等中间层数据也可以对外提供服务，进行实时报表统计。用户可以在中间层进行查询操作，对接各种实时应用、实时大屏。比如

• 直接查DWD层的数据，典型的如根据用户ID返回推荐商品（KV场景）

--Hologres SQL--场景4: 根据用户特征推荐商品SELECT*FROM dws_users 
WHEREuser_id =? AND ds ='2022-11-09’;--场景4: 根据店铺特征推荐商品SELECT * FROM dws_shops WHERE shop_id = ? AND ds = '2022-11-09’;

• 实时报表查看订单量和退单量（OLAP）。

--Hologres SQL--场景6:基于宽表数据展示实时报表-- 最近30天,每个品类的订单总量和退单总量SELECTTO_CHAR(order_create_time,'YYYYMMDD'),order_product_catalog_name,COUNT(*),COUNT(CASE WHEN refund_id ISNOTNULL THEN 1 ELSE null END)FROMdwd_orders
WHEREorder_create_time > now()-'30 day':: inteval
GROUPBY1,2ORDERBY1,2;

第五步，问题排查：Flink数据探查。如果某个业务指标出现异常，Flink可以直接探查每层表的数据来快速定位。比如用Flink探查Hologres DWD层的orders表。Hologres支持Flink的流模式和批模式对数据的探查。

由于流模式是Flink的默认模式，因此我们不需要设置执行模式。它可以直接记录数据变化，从而非常方便的查看数据异常。流模式可以探查获取一段时间范围内的数据及其变化情况。

--Flink SQL-- 流模式探查 SELECT*FROM holo.order_dw.dwd_orders/*+ OPTIONS('cdcMode'='false', 'startTime'='2022-11-09 00:00:00') */ c 
WHERE  user_id =0;

与此同时，批模式探查是获取当前时刻的最新数据。Hologres也支持Flink批模式的数据探查。批模式和流模式的区别在于，流模式关注的是变化，批模式关注的是表中的最新状态。

--Flink SQL-- 批模式探查 set'execution.runtime-mode'='batch’;SELECT   * FROM   holo.order_dw.dwd_orders WHERE   user_id = 0  AND order_create_time>'2022-11-0900:00:00';

五、总结

Hologres跟Flink深度集成。实现完整的Streaming Warehouse方案，该方案有如下明显优势：

1. 一站式：全链路都可以用SQL表示，并且只需要用到Flink和Hologres两个组件，操作非常方便。实时ETL链路、数据分层完全可以用Flink SQL实现，Hologres提供对外提供在线服务和OLAP查询，每层数据可复用、可查，方便构建实时数仓的数据分层和复用体系。
2. 高性能：这种方案可以使得使得Hologres发挥极致的实时写入、实时更新能力和多维OLAP、高并发点查能力，Flink发挥实时加工能力。
3. 企业级：自带多种企业级能力，不仅运维更简单，可观测性更好，安全能力更强，也提供多种高可用能力，从而企业更加方便的构建企业级的Streaming Warehouse。