E-MapReduce:Deltaコネクタ

前提条件

DataLakeクラスターまたはカスタムクラスターが作成され、Trinoサービスが選択されています。 Hadoopクラスターが作成され、Prestoサービスが選択されています。 詳細については、「クラスターの作成」をご参照ください。

制限

EMR V3.39.1、EMR V5.5.0、およびそれ以降のマイナーバージョン、DataLakeクラスターとカスタムクラスターのHadoopクラスターのみがDeltaコネクタをサポートしています。

基本的な使用方法

Deltaコネクタの構成の変更

Deltaコネクタの構成を変更する方法の詳細については、「組み込みコネクタの構成の変更」をご参照ください。

Deltaコネクタのデフォルト構成

例

Trinoを使用してDelta Lakeテーブルを作成または変更することはできません。 Spark SQLを使用してDelta Lakeテーブルを作成できます。 詳細については、「Delta Lakeの使用」をご参照ください。

1. データの生成。

   1. 次のコマンドを実行して、Spark SQL CLIを開きます。

      spark-sql

   2. 次のステートメントを実行して、delta_tableという名前のDelta Lakeテーブルを作成します。

      CREATE TABLE delta_table (id INT) USING delta;

   3. 次のステートメントを実行して、delta_tableテーブルにデータを書き込みます。

      INSERT INTO delta_table VALUES 0,1,2,3,4;

2. データのクエリ。

   1. Trinoコンソールに移動します。 詳細については、「コマンドを実行してTrinoコンソールにログオンする」をご参照ください。

   2. 次のステートメントを実行して、delta_tableテーブルのデータをクエリします。

      SELECT * FROM delta_table;

      次の出力が返されます。

       id
      ----
        0
        1
        2
        3
        4
      (5 rows)

高度な使用方法

タイムトラベル

タイムトラベル機能を使用して、Delta Lakeテーブルの履歴データをクエリできます。

例：

1. 次のコマンドを実行して、Spark SQL CLIを開きます。

   spark-sql

2. 次のステートメントを実行して、データを上書きします。

   INSERT OVERWRITE TABLE delta_table VALUES 5,6,7,8,9;

3. データのクエリ。

   1. Trinoコンソールに移動します。 詳細については、「コマンドを実行してTrinoコンソールにログオンする」をご参照ください。

   2. 次のステートメントを実行して、delta_tableテーブルのデータをクエリします。

      SELECT * FROM delta_table;

      次の出力が返されます。

       id
      ----
        5
        6
        7
        8
        9
      (5 rows)

4. タイムトラベル機能を使用して、delta_tableテーブルの履歴データをクエリします。

   次のステートメントを実行して、バージョン番号でデータをクエリします。 ビジネス要件に基づいて、次のステートメントのVERSIONパラメーターを構成できます。 バージョン番号は、単調に増加する整数です。 デフォルトでは、最初のINSERT操作後にVERSIONパラメーターの値は 1 になり、変更が実行されるたびにこのパラメーターの値は 1 ずつ増加します。

   SELECT * FROM delta_table FOR VERSION AS OF 1;

   次の出力が返されます。

    id
   ----
     2
     1
     3
     4
     0
   (5 rows)

   タイムスタンプでDelta Lakeテーブルの履歴データをクエリすることもできます。 サポートされているタイムスタンプタイプは、DATE、TIMESTAMP、およびTIMESTAMP WITH TIME ZONEです。

   • DATEタイプのタイムスタンプに基づいてデータをクエリする場合、クエリ日の協定世界時（UTC）形式でタイムスタンプが00:00:00のデータがクエリされます。

   • TIMESTAMPタイプのタイムスタンプに基づいてデータをクエリする場合、UTC形式のタイムスタンプが指定されたタイムスタンプに対応するデータがクエリされます。

     たとえば、TIMESTAMPタイプに基づいて、2022年 2 月 15 日 20:00:00 UTC+8のデータをクエリします。 サンプルコード：

     SELECT * FROM delta_table FOR TIMESTAMP AS OF TIMESTAMP '2022-02-15 12:00:00';

     説明

     サンプルコードでは、最初のTIMESTAMPはタイムトラベル機能にタイムスタンプモードが使用されていることを示し、2番目のTIMESTAMPはTIMESTAMPタイプを使用してデータがクエリされていることを示します。

     次の出力が返されます。

      id
     ----
       2
       0
       3
       4
       1
     (5 rows)

   • TIMESTAMP WITH TIME ZONEタイプのタイムスタンプに基づいてデータをクエリする場合、データ型はデータが読み取られる前に暗黙的に変換されます。

     たとえば、TIMESTAMP WITH TIME ZONEタイプに基づいて、2022年 2 月 15 日 20:00:00 UTC+8のデータをクエリします。 サンプルコード：

     SELECT * FROM delta_table FOR TIMESTAMP AS OF CAST('2022-02-15 20:00:00 +0800' AS TIMESTAMP WITH TIME ZONE);

Zオーダー

Trinoは、Zオーダーに基づいてDelta Lakeテーブルのクエリを最適化します。 Parquetに基づくデータクエリの最適化とデータスキップがサポートされています。 Parquetまたはデータスキップに基づいてデータクエリを最適化すると、Delta Lakeはファイルの粒度で各フィールドの最大値と最小値に関する統計を収集します。 統計は、データファイルをフィルタリングするために使用されます。 EMR Trinoによって提供されるDeltaコネクタを使用して、統計を読み取ることができます。

OPTIMIZEステートメントとZORDER BY句を使用して、Delta Lakeテーブルを最適化し、Zオーダーに適切な列を指定できます。 このようにして、Trinoを使用して最適化されたDelta Lakeテーブルをクエリすると、クエリ速度を最大数十倍向上させることができます。

Trinoでは、INT、LONG、DOUBLE、FLOAT、BINARY、BOOLEAN、STRING、およびARRAYのデータ型の列をZオーダーできます。

たとえば、conn_zorderテーブルには、src_ip、src_port、dst_ip、dst_portの 4 つの列が含まれています。

OPTIMIZE conn_zorder ZORDER BY (src_ip, src_port, dst_ip, dst_port);