Hologres:Hologres と Delta Lake を使用したデータレイクハウスの構築

背景情報

• Delta Lake は Databricks 社が提供するデータレイクソリューションです。Delta Lake はデータ中心であり、データインジェスト、データ管理からデータクエリ、データエグレスまで、データストリームのための一連の機能を提供します。これにより、サードパーティの上流および下流のツールを使用して、高速で使いやすく、安全なデータレイクを構築できます。詳細については、「Delta Lake」をご参照ください。

• E-MapReduce (EMR) は、Alibaba Cloud が提供するクラウドネイティブなオープンソースのビッグデータプラットフォームです。EMR は、Hadoop、Hive、Spark、Flink などのオープンソースのビッグデータコンピューティングエンジンとストレージエンジンを提供し、簡単に統合できます。これにより、Hadoop および Spark エコシステムの他のシステムを使用してデータを分析および処理できます。詳細については、「E-MapReduce とは」をご参照ください。

• Data Lake Formation (DLF) は、クラウド上にデータレイクとデータレイクハウスを構築するのに役立つフルマネージドサービスです。DLF は、統一されたメタデータ管理、統一された権限とセキュリティの管理、便利なデータインジェスト、ワンクリックでのデータ探索を提供します。詳細については、「DLF 製品紹介」をご参照ください。

• Hologres は、DLF と EMR とシームレスに統合されるワンストップのリアルタイムデータウェアハウスです。この統合により、データレイクとデータウェアハウスの間の障壁が取り払われ、完全なデータレイクハウスソリューションが構築されます。データレイクの柔軟性と豊富なエコシステムを、リアルタイムデータウェアハウスのパフォーマンス専有型で複雑なオンライン分析およびエンタープライズグレードの機能と組み合わせます。詳細については、「DLF を使用した OSS ベースのデータレイク内のデータのクエリの高速化」をご参照ください。

全体アーキテクチャ

このソリューションでは、EMR Spark を使用してデータの変換と処理を行います。メタデータは DLF に保存され、データは Object Storage Service (OSS) に保存されます。Hologres は DLF のメタデータ管理機能を使用して、Hudi、Delta、CSV、Parquet、ORC、SequenceFile などのさまざまな形式のデータレイクファイルのクエリを高速化します。これにより、統合されたデータレイクハウス分析が可能になります。データは、ビジネスインテリジェンス (BI) レポート、データ可視化ダッシュボード、および上流アプリケーションで使用できます。

環境の準備

利用手順

Hologres のデータレイクアクセラレーション機能は、次の 2 つのシナリオをサポートしています。必要に応じてシナリオを選択してください。

シナリオ 1：OSS 内のテーブルデータに対するクエリを直接高速化する

例:

-- dlf_fdw 拡張機能を作成します。

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

シナリオ 2：クエリパフォーマンスを向上させるために、データを Hologres 標準ストレージにインポートする

Hologres 標準ストレージは SSD (NVMe) ディスクを使用しており、より高いランダム読み書き性能を実現します。OSS 外部テーブルから Hologres 標準ストレージの内部テーブルにデータをインポートできます。その後、インデックスの作成、適切な Shard Count の設定、および適切な分布列の選択によって、クエリパフォーマンスを最適化できます。クエリ 2 (Q2) では、これによりパフォーマンスが 18 倍以上向上する可能性があります。詳細については、「内部テーブルのパフォーマンスの最適化」をご参照ください。

• Hologres に同じ構造の内部テーブルを作成し、データをインポートします。

  次のコードは例です。その他の CREATE TABLE 文については、「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 コアの専用型インスタンスの場合、Hologres 内部テーブルに対するクエリは、OSS 外部テーブルに対するクエリよりも約 100 倍高速です。

• OSS外部テーブル

  • クエリ時間：17.24 秒。

  • 実行プラン:

• Hologres内部テーブル

  • クエリ時間: 106.67ミリ秒

  • 実行プラン: