Data Lake Formation:Paimon プライマリキー テーブル

データ分散

デフォルトでは、各レコードはプライマリキーのハッシュ値に基づいてバケットに割り当てられます。

データ可視性

遅延バケット化テーブルのデータ可視性は、書き込み方法によって異なります。DLF は、以下の条件が満たされた場合に固定バケットへのバッチ書き込みを有効にし、コミット直後にデータが可視になるようにします。

• 削除ベクトルが有効になっておらず、データが Flink バッチジョブまたは INSERT INTO もしくは INSERT OVERWRITE を使用する Spark ジョブによって書き込まれる場合。

• 削除ベクトルが有効になっており、データが Flink バッチジョブまたは INSERT OVERWRITE を使用する Spark ジョブによって書き込まれる場合。

それ以外の場合、データは DLF によって処理された後に可視になり、通常は 1～3 分かかります。DLF がリソースやバケット数の調整を行っている場合、可視になるまでに 5～10 分かかることがあります。特定のストレージ最適化モードを選択して、調整の頻度を減らすことができます。詳細については、「ストレージ最適化モード」をご参照ください。

データ一貫性

• 単一ジョブの書き込み：同じジョブによって書き込まれたデータの場合、DLF は書き込み順序でレコードを処理し、シーケンシャル一貫性を提供します。

  説明

  並列度が変更されていないチェックポイントまたはセーブポイントから再開されたジョブは、同じジョブとして扱われます。

• 複数ジョブの書き込み：異なるジョブによって書き込まれたデータの場合、最終的なテーブルの状態には両方のジョブの結果が混在する可能性がありますが、データは失われず、スナップショット分離の一貫性が提供されます。マージの順序を制御するには、シーケンスフィールドを設定します。詳細については、「順序が乱れたデータの処理」をご参照ください。

データ分散

遅延バケット化と同じです。詳細については、「データ分散」をご参照ください。

データ可視性

可視性は、削除ベクトルが有効かどうかによって異なります。

• 削除ベクトルなし：データはコミット直後に可視になります。

• 削除ベクトルあり：データは DLF 最適化ジョブによって処理された後に可視になり、通常は 1～3 分かかります。

データ分散

動的バケット化テーブルは、既存のバケットにデータを書き込み、バケットが容量を超えると自動的に新しいバケットを作成します。以下のプロパティが動作を制御します。

動的バケットが設定されている場合、更新のパフォーマンスとリソース使用量は、プライマリキーがパーティションキーを完全に含むかどうかによって異なります。

• プライマリキーがパーティションキーを含む場合：更新は同じパーティション内にとどまり、ヒープメモリを使用します。

• プライマリキーがパーティションキーを完全に含まない場合：更新はパーティションをまたぐ可能性があり、RocksDB を必要とし、パフォーマンスが低下する可能性があります。

データ可視性

固定バケットと同じです。詳細については、「データ可視性」をご参照ください。

サポートされる集計関数

• sum: 合計を算出します。型: DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE。

• product: 積を計算します。型: DECIMAL、TINYINT、SMALLINT、INTEGER、BIGINT、FLOAT、DOUBLE。

• max: 最大値を返します。型: CHAR, VARCHAR, DECIMAL, TINYINT, SMALLINT, INTEGER, BIGINT, FLOAT, DOUBLE, DATE, TIME, TIMESTAMP, TIMESTAMP_LTZ。

• min: 最小値を返します。max と同じ型です。

• first_value: 最初の入力値 (null を含む) を返します。すべての型。

• first_not_null_value: 最初の NULL でない入力値を返します。すべての型。

• last_value: 最新の入力値 (null を含む) を返します。すべての型に対応します。

• last_non_null_value (デフォルト)：最新の非 null の入力値を返します。すべての型。

• listagg: 文字列をカンマで連結します。型: STRING。

• bool_and / bool_or: 論理 AND / 論理 OR。型: BOOLEAN。

• rbm32 / rbm64: 32 ビット / 64 ビットの RoaringBitmaps をマージします。 型: VARBINARY。

• nested_update: 行を配列に集約します。'fields.<field-name>.nested-key' = 'pk0,pk1,...' を設定して、ネストされたキーによって重複を排除します (最後の項目が保持されます)。

• nested_partial_update: nested_update と同様ですが、ネストされたキーによって ARRAY 内で列の部分的な更新を実行します。

• collect: 複数の ARRAY を 1 つに結合します。'fields.<field-name>.distinct' = 'true' を設定すると、重複を排除できます。

• merge_map：複数の MAP をマージし、重複するキーについては最後の値を保持します。

例

複数のサブオーダーを集約するには、nested_update を使用します:

-- メインの orders テーブル
CREATE TABLE orders (
  order_id BIGINT PRIMARY KEY NOT ENFORCED,
  user_name STRING,
  address STRING
);

-- サブオーダーテーブル
CREATE TABLE sub_orders (
  order_id BIGINT,
  sub_order_id INT,
  product_name STRING,
  price BIGINT,
  PRIMARY KEY (order_id, sub_order_id) NOT ENFORCED
);

-- ワイドテーブル
-- サブオーダーを集計
-- sub_order_id に基づいてサブオーダーを重複排除
CREATE TABLE order_wide (
  order_id BIGINT PRIMARY KEY NOT ENFORCED,
  user_name STRING,
  address STRING,
  sub_orders ARRAY<ROW<sub_order_id BIGINT, product_name STRING, price BIGINT>>
) WITH (
  'merge-engine' = 'aggregation',
  'fields.sub_orders.aggregate-function' = 'nested_update',
  'fields.sub_orders.nested-key' = 'sub_order_id'
);

INSERT INTO order_wide

SELECT
  order_id,
  user_name,
  address,
  CAST (NULL AS ARRAY<ROW<sub_order_id BIGINT, product_name STRING, price BIGINT>>)
FROM orders

UNION ALL

SELECT
  order_id,
  CAST (NULL AS STRING),
  CAST (NULL AS STRING),
  ARRAY[ROW(sub_order_id, product_name, price)]
FROM sub_orders;

orders テーブルにデータがあるとします:

• (1, 'Alice', 'add1'),

• (2, 'Bob', 'add2')

sub_orders テーブルにはデータがあります：

• (1, 11, 'apple', 10),

• (1, 12, 'banana', 20),

• (2, 21, 'cherry', 30),

• (2, 22, 'peach', 40),

• (2, 21, 'cherry', 50)

order_wide テーブルには次のデータが格納されます。

• (1, 'Alice', 'add1', [(11, 'apple', 10), (12, 'banana', 20)])

• (2, 'Bob', 'add2', [(21, 'cherry', 50), (22, 'peach', 40)])

シーケンスグループ

グローバルな シーケンスフィールド に加えて、シーケンスグループを使用して、異なる列グループに対して独立したマージ順序を指定できます。これは、複数のソーステーブルを持つテーブルのワイド化シナリオで役立ちます。

例：

CREATE TABLE T (
  k INT,
  a STRING,
  b STRING,
  g_1 INT,
  c STRING,
  d STRING,
  g_2 INT,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine' = 'partial-update',
  'fields.g_1.sequence-group' = 'a,b',
  'fields.g_2.sequence-group' = 'c,d'
);

列 a と b はマージシーケンスとして g_1 を使用し (値が大きいほどデータが新しいことを意味します)、列 c と d はマージシーケ-ケンスとして g_2 を使用します。

集計によるテーブルのワイド化

部分更新では、fields.<field-name>.aggregate-function を設定することで、特定のカラムに集計関数を適用することもできます。

例：

CREATE TABLE T (
  k INT,
  a STRING,
  b INT,
  g_1 INT,
  c STRING,
  d INT,
  g_2 INT,
  PRIMARY KEY (k) NOT ENFORCED
) WITH (
  'merge-engine' = 'partial-update',
  'fields.g_1.sequence-group' = 'a,b',
  'fields.b.aggregate-function' = 'max',
  'fields.g_2.sequence-group' = 'c,d',
  'fields.d.aggregate-function' = 'sum'
);

列 a と b は、マージシーケンスとして g_1 を使用します (値が大きいほどデータが新しいことを意味します)。a は最新の null でない値 (デフォルトは last_non_null_value) を保持し、b は最大値を保持します。列 c と d は、マージシーケンスとして g_2 を使用します。c は最新の null でない値を保持し、d は合計を計算します。

不要なチェンジログの削減

デフォルトでは、Paimon は更新された値が以前の値と同一である場合でもチェンジログを生成します。'changelog-producer.row-deduplicate' = 'true' を設定して、このような no-op チェンジログを排除します。これはルックアップメカニズムにのみ適用されます。

no-op チェンジログが頻繁に発生する場合にのみ使用してください。追加の比較オーバーヘッドが発生するためです。