E-MapReduce:Delta コネクタ

前提条件

DataLake クラスターまたは Trino サービスを含むカスタムクラスター、もしくは Presto サービスを含む Hadoop クラスターのいずれかを作成済みである必要があります。詳細については、「クラスターの作成」をご参照ください。

制限事項

Delta コネクタは、EMR-3.39.1 以降または EMR-5.5.0 以降を実行している DataLake クラスター、カスタムクラスター、および Hadoop クラスターでサポートされています。

基本的な使用方法

コネクタ構成の変更

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

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

例

Trino では Delta Lake テーブルの作成や変更はできません。これらの操作は代わりに Spark SQL を使用してください。詳細については、「基本的な使用方法」をご参照ください。

1. データを生成します。

   1. 次のコマンドを実行して Spark SQL コマンドラインに入ります。

      spark-sql

   2. 次の文を実行して Delta Lake テーブルを作成します。

      CREATE TABLE delta_table (id INT) USING delta;

   3. 次の文を実行してテーブルにデータを書き込みます。

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

2. データをクエリします。

   1. Trino コマンドラインにアクセスします。詳細については、「コマンドラインインターフェイスを使用した Trino へのアクセス」をご参照ください。

   2. 次の文を実行してテーブルをクエリします。

      SELECT * FROM delta_table;

      次の出力が返されます。

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

高度な使用方法

タイムトラベル

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

例：

1. 次のコマンドを実行して Spark SQL コマンドラインに入ります。

   spark-sql

2. 次の文を実行してデータを上書きします。

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

3. データをクエリします。

   1. Trino コマンドラインにアクセスします。詳細については、「コマンドラインインターフェイスを使用した Trino へのアクセス」をご参照ください。

   2. 次の文を実行してテーブルをクエリします。

      SELECT * FROM delta_table;

      次の出力が返されます。

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

4. タイムトラベルを使用して過去のデータをクエリします。

   次の文を実行して、バージョン番号でデータをクエリします。バージョン番号は単調増加する整数です。デフォルトでは、最初の INSERT 操作後にバージョン番号は 1 となり、その後の各変更で 1 ずつ増分されます。

   SELECT * FROM delta_table FOR VERSION AS OF 1;

   次の出力が返されます。

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

   タイムスタンプによるデータのクエリも可能です。サポートされるタイムスタンプの型は、DATE、TIMESTAMP、および TIMESTAMP WITH TIME ZONE の 3 種類です。

   • DATE：指定された日付の協定世界時 (UTC) 00:00:00 のデータをクエリします。

   • TIMESTAMP：UTC の指定されたタイムスタンプのデータをクエリします。

     たとえば、北京時間 (UTC+08:00) 2022 年 2 月 15 日 20:00:00 のデータを TIMESTAMP 型でクエリするには、次の文を実行します。

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

     説明

     この文では、最初の TIMESTAMP キーワードがタイムトラベルモードをタイムスタンプ（バージョン番号モードではない）に設定します。2 番目の TIMESTAMP キーワードは文字列リテラルを TIMESTAMP データの型（DATE 型ではない）にキャストします。

     次の出力が返されます。

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

   • TIMESTAMP WITH TIME ZONE：TIMESTAMP WITH TIME ZONE を使用したクエリには型キャストが必要です。

     たとえば、北京時間 (UTC+08:00) 2022 年 2 月 15 日 20:00:00 のデータをクエリするには、次の文を使用します。

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

Z-order

Trino は、Parquet のネイティブ最適化とデータスキップを活用して Delta Lake テーブルのクエリを最適化するために Z-Ordering を使用します。最適化後、Delta Lake は各列の最小値および最大値などのファイルレベルの統計情報を収集し、Delta コネクタはこれらの情報を使用してデータファイルを直接フィルターします。

OPTIMIZE および ZORDER BY コマンドで最適化された Delta Lake テーブルでは、適切に構成された Z-Ordering 列により、Trino クエリが数十倍高速になります。

Trino は、Int、Long、Double、Float、Binary、Boolean、String、および Array のデータの型に対して Z-Ordering をサポートしています。

たとえば、src_ip、src_port、dst_ip、dst_port の 4 つの列を持つ conn_zorder という名前のテーブルを考えてみます。

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