Hologres基于Delta Lake实现湖仓一体

背景信息

• Delta Lake是DataBricks公司推出的一种数据湖方案。Delta Lake以数据为中心，围绕数据流走向（数据从流入数据湖、数据组织管理和数据查询到流出数据湖）推出了一系列功能特性，协助您搭配第三方上下游工具，搭建快捷、易用和安全的数据湖。详情请参见Delta Lake概述。

• EMR是阿里云提供的云原生开源大数据平台，向您提供简单易集成的Hadoop、Hive、Spark、Flink等开源大数据计算和存储引擎，便于您使用Hadoop和Spark生态系统中的其他周边系统分析和处理数据。详情请参见什么是E-MapReduce。

• DLF是一款全托管的帮助您构建云上数据湖及Lakehouse的服务，为您提供了统一的元数据管理、统一的权限与安全管理、便捷的数据入湖能力以及一键式数据探索能力。详情请参见DLF产品简介。

• Hologres作为一站式实时数仓，与DLF、EMR无缝集成，打破数据湖与数据仓库割裂的体系，构建完整的湖仓一体解决方案，将数据湖的灵活性、生态丰富性与实时数仓的高性能在线复杂分析、企业级能力相结合，为您提供一站式实时湖仓解决方案。详情请参见OSS数据湖加速。

整体架构

本解决方案通过EMR Spark来进行数据加工与处理，元数据存储在DLF中，数据存储在OSS上，Hologres可以利用DLF对OSS元数据的管理能力，对OSS多种格式的湖数据（Hudi、Delta、CSV、Parquet、ORC、SequenceFile）进行加速查询和湖仓融合分析，将数据提供给BI报表、可视化大屏和上层应用进一步消费使用。

环境准备

使用说明

Hologres的数据湖加速能力，可以满足实际业务中以下两种使用场景，您可以根据业务需要选择合适的场景。

场景一：使用Hologres直接加速查询OSS上的表数据

示例：

-- 创建DLF外部表插件

CREATE EXTENSION IF NOT EXISTS dlf_fdw;

-- 创建外部服务器

CREATE SERVER IF NOT EXISTS dlf_server FOREIGN data wrapper dlf_fdw options 
(
    dlf_region 'cn-beijing',
    dlf_endpoint 'dlf-share.cn-beijing.aliyuncs.com',
    oss_endpoint 'oss-cn-beijing-internal.aliyuncs.com'
);

-- 导入外部表定义

IMPORT FOREIGN SCHEMA "test_spark_delta" LIMIT TO 
(
		customer_delta,
		lineitem_delta,
		nation_delta,
		orders_delta,
		part_delta,
		partsupp_delta,
		region_delta,
		supplier_delta
)
FROM SERVER dlf_server INTO oss_ext_tables options (if_table_exist 'update');

-- 查询表数据，以Q22为例

SELECT
        cntrycode,
        count(*) AS numcust,
        sum(c_acctbal) AS totacctbal
FROM
        (
                SELECT
                        substring(c_phone FROM 1 FOR 2) AS cntrycode,
                        c_acctbal
                FROM
                        customer_delta
                WHERE
                        substring(c_phone FROM 1 FOR 2) IN
                                ('24', '32', '17', '18', '12', '14', '22')
                        AND c_acctbal > (
                                SELECT
                                        avg(c_acctbal)
                                FROM
                                        customer_delta
                                WHERE
                                        c_acctbal > 0.00
                                        AND substring(c_phone FROM 1 FOR 2) IN
                                                ('24', '32', '17', '18', '12', '14', '22')
                        )
                        AND NOT EXISTS (
                                SELECT
                                        *
                                FROM
                                        orders_delta
                                WHERE
                                        o_custkey = c_custkey
                        )
        ) AS custsale
GROUP BY
        cntrycode
ORDER BY
        cntrycode;

返回结果：

+------------+-------------+---------------+
| cntrycode  | numcust     | totacctbal    | 
+------------+-------------+---------------+
| 12         | 90805       | 681136537.68  |
| 14         | 91459       | 685826271.21  |
| 17         | 91313       | 685025263.11  |
| 18         | 91292       | 684588251.63  |
| 22         | 90399       | 677402363.79  |
| 24         | 90635       | 680033065.67  |
| 32         | 90668       | 680459221.16  |
+------------+-------------+---------------+

场景二：导入Hologres标准存储以获取更好的查询性能

Hologres标准存储采用SSD（NVME）硬盘，随机读写性能更好。将OSS外表导入Hologres内部标准存储，可通过创建索引、设置适合的Shard数、选择合适的分布列等手段优化查询性能，以Q2为例，可获得18倍以上的性能提升。详情请参见优化内部表性能化。

• Hologres中创建相同结构的内部表并导入数据。

  示例如下。

  --创建内部表
  
  BEGIN;
  CREATE TABLE region
  (
      R_REGIONKEY INT  NOT NULL PRIMARY KEY,
      R_NAME      TEXT NOT NULL,
      R_COMMENT   TEXT
  );
  CALL set_table_property('region', 'distribution_key', 'R_REGIONKEY');
  CALL set_table_property('region', 'bitmap_columns', 'R_REGIONKEY,R_NAME,R_COMMENT');
  CALL set_table_property('region', 'dictionary_encoding_columns', 'R_NAME,R_COMMENT');
  CALL set_table_property('region', 'time_to_live_in_seconds', '31536000');
  COMMIT;
  
  --导入数据
  
  INSERT INTO public.region SELECT * FROM region_delta ;
  

• 内表查询结果。

  SELECT
          cntrycode,
          count(*) AS numcust,
          sum(c_acctbal) AS totacctbal
  FROM
          (
                  SELECT
                          substring(c_phone FROM 1 FOR 2) AS cntrycode,
                          c_acctbal
                  FROM
                          customer
                  WHERE
                          substring(c_phone FROM 1 FOR 2) IN
                                  ('24', '32', '17', '18', '12', '14', '22')
                          AND c_acctbal > (
                                  SELECT
                                          avg(c_acctbal)
                                  FROM
                                          customer
                                  WHERE
                                          c_acctbal > 0.00
                                          AND substring(c_phone FROM 1 FOR 2) IN
                                                  ('24', '32', '17', '18', '12', '14', '22')
                          )
                          AND NOT EXISTS (
                                  SELECT
                                          *
                                  FROM
                                          orders
                                  WHERE
                                          o_custkey = c_custkey
                          )
          ) AS custsale
  GROUP BY
          cntrycode
  ORDER BY
          cntrycode;

性能对比

以32 Core独享实例为例，可以看到Hologres内部表查询速度比OSS外部表会高出约100倍：

• OSS外部表

  • 查询耗时：17.24s。

  • 执行计划：

• Hologres内部表

  • 查询耗时：106.67ms。

  • 执行计划：