Delta テーブルのパーティション分割

Hive スタイルのパーティション分割では、1 つ以上のパーティション列の値に基づいて、Delta テーブルが物理サブディレクトリに分割されます。 パーティション列の値の一意の組み合わせごとに、個別のディレクトリが作成されます。 このレイアウトにより、パーティションの排除が有効になります。クエリがパーティション列にフィルターを適用すると、エンジンはディレクトリ全体をスキップします。

パーティション分割を使用する場合

Delta Lake でのパーティション分割の主なユース ケースは、 競合しない同時書き込み操作を有効にすることです。 Delta Lake では オプティミスティック コンカレンシー制御が使用され、同じファイルにアクセスする 2 つの操作が競合する可能性があります。 パーティション分割を使用すると、個別のパーティションで操作することで、同時操作で不整合なファイル セットをターゲットにできます。

パーティション分割は、次の場合に使用します。

• 複数のパイプラインが異なる部署やリージョンを同時に処理するなど、競合することなく、同じテーブルに更新、削除、またはマージする必要がある 同時実行ライター があります。
• パーティション列の カーディナリティは低から中程度 です (数万から数百の個別の値ではなく、数十から数百)。 注: 大きなテーブルでは、より多くのパーティションに対応できます。 各パーティションで少なくとも 1 GB のデータをターゲットにします。
• パーティション値は 書き込みパターンに合わせて調整されます。各ライターは、自然に特定のパーティションを対象とします。

パーティション テーブルを作成します。

CREATE TABLE sales.orders (
    order_id BIGINT,
    order_date DATE,
    region STRING,
    amount DECIMAL(10,2)
)
USING DELTA
PARTITIONED BY (region)

from delta.tables import DeltaTable

DeltaTable.create(spark) \
    .tableName("sales.orders") \
    .addColumn("order_id", "BIGINT") \
    .addColumn("order_date", "DATE") \
    .addColumn("region", "STRING") \
    .addColumn("amount", "DECIMAL(10,2)") \
    .partitionedBy("region") \
    .execute()

import io.delta.tables.DeltaTable

DeltaTable.create(spark)
    .tableName("sales.orders")
    .addColumn("order_id", "BIGINT")
    .addColumn("order_date", "DATE")
    .addColumn("region", "STRING")
    .addColumn("amount", "DECIMAL(10,2)")
    .partitionedBy("region")
    .execute()

パーティション分割と同時書き込み

パーティション分割は、同時書き込み操作間の競合を回避するための Delta Lake の主要なメカニズムです。 テーブルがパーティション分割されている場合、異なるパーティションを対象とする操作は、不整合なファイル セットに対して動作し、相互に競合しません。

たとえば、パーティション列がマージ条件MERGE INTO場合、regionによってパーティション分割されたテーブルに対する 2 つの同時操作は、それぞれが異なるリージョンをターゲットにしている限り競合しません。

-- Pipeline A: processes North America only
MERGE INTO sales.orders AS target
USING staged_orders AS source
ON target.order_id = source.order_id
    AND target.region = 'NA'
    AND source.region = 'NA'
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *

spark.sql("""
MERGE INTO sales.orders AS target
USING staged_orders AS source
ON target.order_id = source.order_id
    AND target.region = 'NA'
    AND source.region = 'NA'
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
""")

spark.sql("""
MERGE INTO sales.orders AS target
USING staged_orders AS source
ON target.order_id = source.order_id
    AND target.region = 'NA'
    AND source.region = 'NA'
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *
""")

パーティション分割を行わない場合、または操作条件にパーティション列を含めなくても、論理的に異なる行を変更した場合でも、これらの同じ操作が競合する可能性があります。 パーティション列は、ソース データだけでなく、マージ条件自体に含まれている必要があります。 これを使用しないと、Delta Lake は検証時に、2 つの操作が不整合なファイル セットに触れたことを判断できません。

競合の種類と解決戦略の完全なガイドについては、 コンカレンシー制御に関するページを参照してください。

よくある落とし穴

• カーディナリティの高い パーティション列 (たとえば、数百万の値を持つ user_id ) では、数千もの小さなディレクトリとファイルが作成され、書き込みと読み取りの両方のパフォーマンスが低下します。
  • 日付列は慎重に選択する必要があります。 多くのテーブルでは、日付列でパーティション分割すると、小さなパーティションが多すぎます。 各パーティションで少なくとも 1 GB のデータをターゲットにします。
• テーブル全体を書き換えないと、テーブルの作成後にパーティション列を変更することはできません。
• 小さなファイルの問題 は、ストリーミングや多数のパーティションへの頻繁な追加で一般的です。各書き込みではパーティションごとに少なくとも 1 つのファイルが作成されるためです。
• パーティション分割と液体クラスタリングは、 同じテーブルに対して互換性がありません。 1 つの戦略を選択する必要があります。

パーティション分割と液体クラスタリングの比較

関連するコンテンツ

• 同時実行制御
• 液体クラスタリング
• Z オーダー
• テーブル圧縮
• ファイルをスキップ
• Vオーダー
• Delta テーブルのメンテナンス