E-MapReduce:Delta Lake および Hudi のデータに Hive を使用してアクセスする

前提条件

• Hive、Delta Lake、および Hudi サービスを含むクラスターが作成されていること。詳細については、「クラスターの作成」をご参照ください。

• クラスターにログオンしていること。詳細については、「クラスターへのログオン」をご参照ください。

制限事項

E-MapReduce（EMR） V3.36.0 以降のマイナーバージョン、または EMR V5.2.0 以降のマイナーバージョンでのみ、Hive を使用して Hudi データを読み取ることができます。

Hive を使用して Delta Lake データにアクセスする

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

   spark-sql

2. Spark でテーブルを作成し、テーブル内のデータをクエリします。

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

      create table delta_table (id int) using delta;

   2. 次のステートメントを実行して、テーブルにデータを挿入します：

      insert into delta_table values 0,1,2,3,4;

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

      select * from delta_table;

      次の出力が返されます。

      2
      3
      4
      0
      1
      Time taken: 1.847 seconds, Fetched 5 row(s)

3. Hive で Delta Lake データをクエリします。

   1. 次のコマンドを実行して、Hive CLI を開きます：

      hive

   2. 次のコマンドを実行して、Hive 内の Delta Lake テーブルの詳細情報を表示します：

      desc formatted delta_table;

   3. 次のコマンドを実行して、Hive 内の Delta Lake テーブルのデータをクエリします：

      select * from delta_table;

      次の出力が返されます。

      OK
      2
      3
      4
      0
      1
      Time taken: 1.897 seconds, Fetched: 5 row(s)

      説明

      上記の情報は、Hive でクエリされたデータが Spark の Delta Lake テーブルのデータと同じであることを示しています。これは、Hive を使用して Delta Lake データにアクセスしたことを示しています。

Hive を使用して Hudi データにアクセスする

Spark を使用して Hudi からデータを読み取り、Hudi にデータを書き込む

• Spark SQL を起動する方法

  • Spark 2 または Spark 3 とバージョン 0.11 より前の Hudi を使用する場合は、次のコードを実行します。

    spark-sql \
    --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
    --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension'

  • Spark 3 と Hudi 0.11 以降を使用する場合は、次のコードを実行します。

    spark-sql \
    --conf 'spark.serializer=org.apache.spark.serializer.KryoSerializer' \
    --conf 'spark.sql.extensions=org.apache.spark.sql.hudi.HoodieSparkSessionExtension' \
    --conf 'spark.sql.catalog.spark_catalog=org.apache.spark.sql.hudi.catalog.HoodieCatalog'

• 例

  • テーブルを作成します。

    create table h0 (
      id bigint,
      name string,
      price double,
      ts long
    ) using hudi
    tblproperties (
      primaryKey="id",
      preCombineField="ts"
    );

  • テーブルの詳細を表示します。

    desc formatted h0;

    次の出力が返されます。

    _hoodie_commit_time     string
    _hoodie_commit_seqno    string
    _hoodie_record_key      string
    _hoodie_partition_path    string
    _hoodie_file_name       string
    id                      bigint
    name                    string
    price                   double
    ts                      bigint

    説明

    デフォルトでは、_hoodie_commit_time、_hoodie_commit_seqno、_hoodie_record_key、_hoodie_partition_path、および _hoodie_file_name フィールドは Hudi によって追加されます。

  • テーブルに対する操作を実行します。

    -- insert  // 挿入
    insert into h0 values (1, 'a1', 10, 1000), (2, 'a2', 11, 1000);
    
    -- update // 更新
    update h0 set name = 'a1_new' where id = 1;
    
    -- delete // 削除
    delete from h0 where id = 1;

  • テーブル内のデータをクエリします。

    • 例 1

      select id, name, price, ts from h0;

      次の出力が返されます。

      2    a2    11.0    1000

    • 例 2

      select * from h0;

      次の出力が返されます。

      4.820221130150621338    20221130150621338_0_1    id:2        40d6507e-0579-42ce-a10f-c5e07a3981e5-0_0-29-2007_2022113015062****.parquet    2    a2    11.0    1000

      説明

      テーブルはパーティション化されていないテーブルです。したがって、_hoodie_partition_path フィールドは空のままです。上記の出力の最初の 4 つのフィールド値は、Hudi によって追加されたデフォルトフィールドの値です。

  Spark を使用して Hudi からデータを読み取り、Hudi にデータを書き込む例については、「Spark ガイド」をご参照ください。

Hive を使用して Hudi データをクエリする

-- select // 選択
select * from h0;