Delta Lake を使用してデータを選択的に上書きする

Delta Lake には、選択的上書き用に次の個別のオプションがあります。

ほとんどのユース ケースでは、Databricks では REPLACE USING または REPLACE WHEREを使用することをお勧めします。 ユース ケースで複雑または NULL セーフな一致条件が必要な場合にのみ、 REPLACE ON を使用します。

各オプションの置換動作の詳細については、 INSERTを参照してください。 DataFrameWriter Delta Lake オプションの完全な一覧については、Delta Lake と Apache Iceberg を参照してください。

Scala と Python では、replaceOn と replaceUsing は、replaceWhere、partitionOverwriteMode、または overwriteSchema と組み合わせて使用することはできません。

空のソース クエリの場合、 REPLACE USING と REPLACE ON の両方でデータが削除されませんが、 REPLACE WHERE データが削除される場合があります。

REPLACE WHERE

任意の式に一致するデータのみを REPLACE WHEREで選択的に上書きできます。

start_dateでパーティション分割されたターゲット テーブルの 1 月のイベントを、replace_data内のデータにアトミックに置き換えるには:

Python

(replace_data.write
  .mode("overwrite")
  .option("replaceWhere", "start_date >= '2017-01-01' AND end_date <= '2017-01-31'")
  .saveAsTable("events")
)

Scala

replace_data.write
  .mode("overwrite")
  .option("replaceWhere", "start_date >= '2017-01-01' AND end_date <= '2017-01-31'")
  .saveAsTable("events")

SQL

INSERT INTO TABLE events REPLACE WHERE start_date >= '2017-01-01' AND end_date <= '2017-01-31' SELECT * FROM replace_data

このサンプル コードは、replace_data のデータを書き出し、そのすべての行が述語と一致することを検証し、overwrite セマンティックを使用してアトミックな置換を実行します。 操作内の値が述語の外側にある場合、この操作は既定でエラーで失敗します。

クラシック コンピューティングでは、この動作を述語の範囲内の値を overwrite し、指定した範囲外のレコードを insert するように変更するには、spark.databricks.delta.replaceWhere.constraintCheck.enabled を false に設定して制約チェックを削除します。

Python

spark.conf.set("spark.databricks.delta.replaceWhere.constraintCheck.enabled", False)

Scala

spark.conf.set("spark.databricks.delta.replaceWhere.constraintCheck.enabled", false)

SQL

SET spark.databricks.delta.replaceWhere.constraintCheck.enabled=false

空のソース クエリの場合、テーブル行 REPLACE WHERE 削除される場合があります。

従来の動作

レガシ replaceWhere は、クラシック コンピューティングでのみ使用できます。 クラシック コンピューティングの概要を参照してください。

replaceWhereの従来の動作を使用する場合、クエリはパーティション列に対してのみ述語と一致するデータを上書きします。 次のコマンドは、dateでパーティション分割されているターゲットテーブルの1月を、df内のデータでアトミックに置き換えます。

Python

(df.write
  .mode("overwrite")
  .option("replaceWhere", "birthDate >= '2017-01-01' AND birthDate <= '2017-01-31'")
  .saveAsTable("people10m")
)

Scala

df.write
  .mode("overwrite")
  .option("replaceWhere", "birthDate >= '2017-01-01' AND birthDate <= '2017-01-31'")
  .saveAsTable("people10m")

従来の動作を使用するには、 spark.databricks.delta.replaceWhere.dataColumns.enabled を false に設定します。

Python

spark.conf.set("spark.databricks.delta.replaceWhere.dataColumns.enabled", False)

Scala

spark.conf.set("spark.databricks.delta.replaceWhere.dataColumns.enabled", false)

SQL

SET spark.databricks.delta.replaceWhere.dataColumns.enabled=false

動的データの上書き

動的データの上書きでは、指定されたキー列またはブール式に一致するデータのみが選択的に上書きされ、その他のすべてのデータは変更されません。 パーティション テーブル、パーティション分割されていないテーブル、および液体クラスタリングを含むテーブルがすべてサポートされています。

動的パーティションの上書きは、動的データ上書き動作のサブセットです。 動的パーティション上書きでは、書き込みによって新しいデータがコミットされる各パーティション内の既存データがすべて置き換えられ、それ以外のパーティションは変更されません。 サポートされるのは、パーティション分割されたテーブルのみです。

REPLACE USING

Databricks Runtime 16.3 以降でサポートされている SQL。 Databricks Runtime 18.2 以降でサポートされているPythonと Scala。 Databricks Runtime 16.3 と 17.1 の動作の違いについては、「 レガシ動作」を参照してください。

REPLACE USING では、Databricks SQL ウェアハウス、サーバーレス コンピューティング、クラシック コンピューティングで動作する、コンピューティングに依存しないアトミック上書き動作が可能になります。 REPLACE USING では、Spark セッション構成を設定する必要はありません。

REPLACE USING は、指定した列が等価比較で等しい場合に行を置き換えます。 その他のデータはすべて変更されません。

REPLACE USINGで動的データの上書きを使用します。

Python

(sourceDataDF.write
  .mode("overwrite")
  .option("replaceUsing", "event_id, start_date")
  .saveAsTable("events")
)

Scala

sourceDataDF.write
  .mode("overwrite")
  .option("replaceUsing", "event_id, start_date")
  .saveAsTable("events")

SQL

INSERT INTO TABLE events
  REPLACE USING (event_id, start_date)
  SELECT * FROM source_data

空のソース クエリの場合、 REPLACE USING はテーブル行を削除しません。

複雑または NULL セーフな一致条件の場合は、代わりに REPLACE ON を使用します。 REPLACE ONを参照してください。

SQL 言語リファレンスの INSERT を参照してください。

従来の動作

Databricks Runtime 16.3 から 17.1 では、 REPLACE USING は従来の動作を使用し、動的パーティションの上書きのみを許可しますが、Databricks Runtime 17.2 以降では動的データの上書きが可能です。

REPLACE USING 従来の動作については、次の制約事項と動作を考慮してください。

• USING句では、テーブルのパーティション列の完全なセットを指定する必要があります。
• 書き込まれたデータが予期されるパーティションのみに接していることを常に検証します。 間違ったパーティション内の 1 つの行によって、誤ってパーティション全体が上書きされる可能性があります。

REPLACE ON

Databricks Runtime 17.1 以降でサポートされている SQL。 Databricks Runtime 18.2 以降でサポートされているPythonと Scala。

REPLACE ON は、指定された列の値が等価比較で等しい場合に行を置き換える REPLACE USING とは異なり、ユーザー定義の条件を満たす場合に行を置き換えます。 REPLACE ON値を等しい値として扱うなど、サポートされていないREPLACE USING一致するロジックが必要な場合は、NULLを使用します。

必要に応じて、 targetAlias オプションを使用して、ターゲット テーブルのエイリアスと、ソース データのエイリアスを指定する .as() または .alias() API を指定します。

SQL 構文については、 INSERTを参照してください。

Python

(sourceDataDF.alias("s")
  .write
  .mode("overwrite")
  .option("targetAlias", "t")
  .option("replaceOn", "s.event_id <=> t.event_id AND s.start_date <=> t.start_date")
  .saveAsTable("events")
)

Scala

sourceDataDF.as("s")
  .write
  .mode("overwrite")
  .option("targetAlias", "t")
  .option("replaceOn", "s.event_id <=> t.event_id AND s.start_date <=> t.start_date")
  .saveAsTable("events")

SQL

INSERT INTO TABLE events AS t
  REPLACE ON (s.event_id <=> t.event_id AND s.start_date <=> t.start_date)
  (SELECT * FROM source_data) AS s

空のソース クエリの場合、 REPLACE ON はテーブル行を削除しません。

動的パーティションが partitionOverwriteMode で上書きされる (レガシ)

Databricks Runtime 11.3 LTS 以降では、SQL で INSERT OVERWRITE するか、 df.write.mode("overwrite")を使用した DataFrame 書き込みのどちらかの上書きモードを使用したパーティション テーブルの動的パーティション上書きがサポートされています。 この種類の上書きは、Databricks SQL ウェアハウスやサーバーレス コンピューティングではなく、クラシック コンピューティングでのみ使用できます。

動的パーティション上書きモードを使用するには、Spark セッション構成 spark.sql.sources.partitionOverwriteMode を dynamicに設定します。 または、 DataFrameWriter オプション partitionOverwriteMode を dynamic に設定することもできます。 存在する場合、セッション構成で定義されているモードは、クエリ固有のオプションによりオーバーライドされます。 spark.sql.sources.partitionOverwriteMode の既定値は static です。

partitionOverwriteMode の使用例を次に示します。

SQL

SET spark.sql.sources.partitionOverwriteMode=dynamic;
INSERT OVERWRITE TABLE default.people10m SELECT * FROM morePeople;

Python

(df.write
  .mode("overwrite")
  .option("partitionOverwriteMode", "dynamic")
  .saveAsTable("default.people10m")
)

Scala

df.write
  .mode("overwrite")
  .option("partitionOverwriteMode", "dynamic")
  .saveAsTable("default.people10m")

partitionOverwriteModeについては、次の制約と動作に留意してください。

• overwriteSchemaを true に設定することはできません。
• 同じpartitionOverwriteMode操作でreplaceWhereとDataFrameWriterの両方を指定することはできません。
• replaceWhere オプションを使用してDataFrameWriter条件を指定した場合、Delta Lake はその条件を適用して、上書きするデータを制御します。 このオプションは、 partitionOverwriteMode セッション レベルの構成よりも優先されます。
• 書き込まれたデータが予期されるパーティションのみに接していることを常に検証します。 間違ったパーティション内の 1 つの行によって、誤ってパーティション全体が上書きされる可能性があります。