E-MapReduce:Hive からの Delta Lake および Hudi データへのアクセス

ステップ 1：Spark での Delta Lake テーブルの作成

1. Spark CLI を開きます。

   spark-sql

2. delta_table という名前の Delta Lake テーブルを作成します。

   create table delta_table (id int) using delta;

3. テーブルにデータを挿入します。

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

4. Spark でテーブルをクエリして、データを確認します。

   select * from delta_table;

   出力は次のようになります。

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

ステップ 2：Hive での Delta Lake データのクエリ

1. Hive CLI を開きます。

   hive

2. （オプション）テーブルのメタデータを確認します。

   desc formatted delta_table;

3. 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 データを正常に読み取れることを確認できます。

ステップ 1：Spark を使用した Hudi データの準備

Hive からクエリする前に、Spark SQL を使用して Hudi テーブルを作成し、データを投入します。

Spark SQL の起動

起動コマンドは、ご利用のクラスター内の Spark および Hudi のバージョンによって異なります。

• Spark 2 または 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'

• 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'

Hudi テーブルの作成およびデータ投入

1. h0 という名前の Hudi テーブルを作成します。

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

2. テーブルスキーマを確認します。

   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

   Hudi はすべてのテーブルに自動的に 5 つのメタデータフィールド（_hoodie_commit_time、_hoodie_commit_seqno、_hoodie_record_key、_hoodie_partition_path、および _hoodie_file_name）を追加します。これらのフィールドは、select * を使用するすべてのクエリ結果に表示されます。

3. データの挿入、更新、および削除を行います。

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

4. データを確認します。更新および削除操作後、id=2 の行のみが残ります。

   説明

   h0 は非パーティション化テーブルであるため、_hoodie_partition_path は空です。出力の先頭 5 つの値は、5 つの Hudi メタデータフィールドに対応しています。

   select id, name, price, ts from h0;

   出力は次のようになります。

   2    a2    11.0    1000

   Hudi のメタデータフィールドを結果に含めるには、次のコマンドを実行します。

   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

Hudi の Spark に関するその他の例については、「Spark Guide」をご参照ください。

ステップ 2：Hive での Hudi データのクエリ

Hive CLI を開くか、Beeline を使用して Hive ThriftServer に接続し、Hudi テーブルに対してクエリを実行します。

-- select
select * from h0;