E-MapReduce:Kudu コネクタ

スキーマエミュレーションが無効な場合 (デフォルト)

スキーマエミュレーションが無効な場合、すべての Kudu テーブルは default スキーマに表示されます。

orders テーブルを完全修飾名でクエリします。

SELECT * FROM kudu.default.orders;

kudu をカタログとして、default をスキーマとして設定した場合、クエリは次のように簡略化されます。

SELECT * FROM orders;

テーブル名に特殊文字が含まれる場合は、二重引用符で囲みます。

SELECT * FROM kudu.default."special.table!";

クイック例：テーブルの作成とクエリ

1. default スキーマに users という名前のテーブルを作成します。

   CREATE TABLE kudu.default.users (
     user_id int WITH (primary_key = true),
     first_name varchar,
     last_name varchar
   ) WITH (
     partition_by_hash_columns = ARRAY['user_id'],
     partition_by_hash_buckets = 2
   );

   テーブルを作成する際は、プライマリキー、列のエンコーディングまたは圧縮形式、およびパーティション情報 (ハッシュまたはレンジ) を指定します。

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

   DESCRIBE kudu.default.users;

   期待される出力：

      Column   |  Type   |                      Extra                      | Comment
   ------------+---------+-------------------------------------------------+---------
    user_id    | integer | primary_key, encoding=auto, compression=default |
    first_name | varchar | nullable, encoding=auto, compression=default    |
    last_name  | varchar | nullable, encoding=auto, compression=default    |
   (3 rows)

3. 行を挿入します。

   INSERT INTO kudu.default.users VALUES (1, 'Donald', 'Duck'), (2, 'Mickey', 'Mouse');

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

   SELECT * FROM kudu.default.users;

スキーマエミュレーションが有効な場合

etc/catalog/kudu.properties で kudu.schema-emulation.enabled=true を設定して、スキーマエミュレーションを有効にします。これにより、Kudu テーブルはその名前に基づいて Trino スキーマにマッピングされます。

空のプレフィックスでのマッピング (kudu.schema-emulation.prefix=)

Kudu はスキーマをサポートしていません。Trino はスキーマを管理するために $schemas という名前の特別なテーブルを作成します。

標準プレフィックスでのマッピング (kudu.schema-emulation.prefix=presto::)

標準プレフィックスを使用する場合、Trino はスキーマを管理するために presto::$schemas という名前の特別なテーブルを作成します。

Trino から Kudu へ

Trino から Kudu にデータを書き込む場合、以下のデータ型マッピングが適用されます。

以下の Trino 型はサポートされていません：TIME、JSON、TIME WITH TIMEZONE、TIMESTAMP WITH TIME ZONE、INTERVAL YEAR TO MONTH、INTERVAL DAY TO SECOND、ARRAY、MAP、IPADDRESS。

列のプロパティ

列のプロパティは WITH 句で指定します。

エンコーディングと圧縮を明示的に指定した例：

CREATE TABLE mytable (
  name varchar WITH (primary_key = true, encoding = 'dictionary', compression = 'snappy'),
  index bigint WITH (nullable = true, encoding = 'runlength', compression = 'lz4'),
  comment varchar WITH (nullable = true, encoding = 'plain', compression = 'default'),
  ...
) WITH (...);

パーティション設計

Kudu はハッシュパーティションとレンジパーティションをサポートしています。テーブルには、どちらかのタイプのパーティションが少なくとも 1 つ必要です。

ハッシュパーティションの定義

partition_by_hash_columns (パーティションキー列) と partition_by_hash_buckets (バケット数) を指定します。パーティションキー列は、プライマリキー列のサブセットである必要があります。

1 つのハッシュパーティショングループ：

CREATE TABLE mytable (
  col1 varchar WITH (primary_key=true),
  col2 varchar WITH (primary_key=true),
  ...
) WITH (
  partition_by_hash_columns = ARRAY['col1', 'col2'],
  partition_by_hash_buckets = 4
);

col1 と col2 がハッシュパーティションキー列です。行は 4 つのバケットに分散されます。

2 つの独立したハッシュパーティショングループ：

CREATE TABLE mytable (
  col1 varchar WITH (primary_key=true),
  col2 varchar WITH (primary_key=true),
  ...
) WITH (
  partition_by_hash_columns = ARRAY['col1'],
  partition_by_hash_buckets = 2,
  partition_by_second_hash_columns = ARRAY['col2'],
  partition_by_second_hash_buckets = 3
);

最初のグループは col1 によって行を 2 つのバケットに分散し、2 番目のグループは col2 によって 3 つのバケットに分散します。パーティションの総数は 6 (2 x 3) です。

レンジパーティションの定義

partition_by_range_columns を使用してレンジパーティション列を指定し、range_partitions を使用して初期のパーティション境界を定義します。

CREATE TABLE events (
  rack varchar WITH (primary_key=true),
  machine varchar WITH (primary_key=true),
  event_time timestamp WITH (primary_key=true),
  ...
) WITH (
  partition_by_hash_columns = ARRAY['rack'],
  partition_by_hash_buckets = 2,
  partition_by_second_hash_columns = ARRAY['machine'],
  partition_by_second_hash_buckets = 3,
  partition_by_range_columns = ARRAY['event_time'],
  range_partitions = '[{"lower": null, "upper": "2018-01-01T00:00:00"},
                       {"lower": "2018-01-01T00:00:00", "upper": null}]'
);

このテーブルには 2 つのハッシュパーティショングループと、event_time に対する 1 つのレンジパーティションがあり、2018-01-01T00:00:00 で分割されます。

レンジパーティションの管理

ストアドプロシージャを使用して、既存のテーブルのレンジパーティションを追加または削除します。

-- レンジパーティションの追加
CALL kudu.system.add_range_partition(<schema_name>, <table_name>, <range_partition_as_json_string>)

-- レンジパーティションの削除
CALL kudu.system.drop_range_partition(<schema_name>, <table_name>, <range_partition_as_json_string>)

パラメーター：

データ型ごとの JSON 値の形式：

例： myschema スキーマの events テーブルに、2018-01-01 から 2018-06-01 までのレコードをカバーするレンジパーティションを追加します。

CALL kudu.system.add_range_partition('myschema', 'events', '{"lower": "2018-01-01", "upper": "2018-06-01"}')

下限の "2018-01-01" は 2018-01-01T00:00:00.000 として解釈されます。

テーブル上の既存のすべてのレンジパーティションを表示するには、SHOW CREATE TABLE を実行します。出力の range_partitions プロパティに、現在のパーティション境界がリストされます。