Hologres:Use Hologres to implement a data lakehouse solution based on Delta Lake

Background information

• Delta Lake is a data lake solution that is developed by Databricks. Delta Lake provides features that can be used to write data to data lakes, manage data, query data, and read data from data lakes. You can build easy-to-use and secure data lakes by using Delta Lake and third-party upstream and downstream tools. For more information, see Overview.

• E-MapReduce (EMR) is a cloud-native open source big data platform provided by Alibaba Cloud. EMR provides you with easy-to-integrate open source big data computing and storage engines, such as Apache Hadoop, Apache Hive, Apache Spark, and Apache Flink. You can easily use other peripheral systems in Hadoop and Spark ecosystems to analyze and process your data. For more information, see What is EMR on ECS?

• Data Lake Formation (DLF) is a fully managed service that is developed by Alibaba Cloud and can help you build cloud-based data lakes and data lakehouses. DLF provides you with centralized metadata management and centralized permission and security management. DLF allows you to ingest data into data lakes with ease and supports push-button data exploration. For more information, see Overview.

• Hologres is a one-stop real-time data warehousing service developed by Alibaba Cloud. Hologres is seamlessly integrated with DLF and EMR and combines data lakes and data warehouses to build a complete data lakehouse solution. The flexibility and diverse ecosystems of data lakes are integrated with the high-performance online complex analytics capabilities and other enterprise-class capabilities of Hologres to provide you with a one-stop real-time data lakehouse solution. For more information, see Use DLF to read data from and write data to OSS.

Architecture

In this solution, EMR Spark is used to process data. Metadata is stored in DLF, and data is stored in Object Storage Service (OSS). Hologres can use the OSS metadata management capability of DLF to accelerate queries of OSS data in multiple formats and analyze such data based on the integration of data lakes and warehouses. The data formats include Apache Hudi, Delta Lake, CSV, Parquet, Optimized Row Columnar (ORC), and SequenceFile. Hologres provides data for business intelligence (BI) reports, dashboards, and upper-layer applications.

Environment preparation

Usage notes

The data lake query acceleration capability of Hologres can be used in the following query acceleration methods. You can select a method based on your business requirements.

Method 1: Use Hologres to accelerate queries of table data stored in OSS

Example:

-- Install the dlf_fdw extension.

CREATE EXTENSION IF NOT EXISTS dlf_fdw;

-- Create a foreign server.

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 the schema of the foreign table.

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');

-- Query the table data. In this example, the TPC-H Q22 query statement is used.

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;

The following result is returned:

+------------+-------------+---------------+
| 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  |
+------------+-------------+---------------+

Method 2: Import data from an OSS foreign table to a Hologres internal table that uses the standard storage mode to improve query performance

In standard storage mode of Hologres, Non-Volatile Memory Express (NVMe) SSDs are used. This storage mode provides better random read and write performance. After you import data from an OSS foreign table to a Hologres internal table that uses the standard storage mode, you can improve the query performance by configuring indexes, setting appropriate shard counts, or selecting an appropriate distribution key. If you use the TPC-H Q2 query statement, the query performance can be improved by more than 18 times. For more information, see Optimize query performance.

• Create a Hologres internal table that has the same schema as the OSS foreign table and import the data from the OSS foreign table to the Hologres internal table.

  The following sample code provides an example. For more information about the statements that are used to create tables, see Query data in Hologres.

  -- Create a Hologres internal table.
  
  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;
  
  -- Import data to the Hologres internal table.
  
  INSERT INTO public.region SELECT * FROM region_delta ;
  

• Execute the following statement to query data from the Hologres internal table:

  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;

Performance comparison

In this example, an exclusive instance with 32 CPU cores is used. The query speed on a Hologres internal table is about 100 times faster than that on an OSS foreign table.

• OSS foreign table

  • Query duration: 17.24s.

  • Execution plan:

• Hologres internal table

  • Query duration: 106.67 ms.

  • Execution plan: