Delta Lake テーブル スキーマの更新

Delta Lakeを使用すると、テーブルのスキーマを更新できます。次の種類の変更がサポートされています：

• 新しい列の追加（任意の位置）
• 既存の列の並べ替え
• 既存の列の名前を変更する

これらの変更は、DDLを使用して明示的に行うことも、DMLを使用して暗黙的に行うこともできます。

important

Deltaテーブルスキーマの更新は、すべての並列Delta書き込み操作と競合する操作です。

Delta テーブル スキーマを更新すると、そのテーブルから読み取ったストリームは終了します。 ストリームを続行する場合は、ストリームを再起動する必要があります。 推奨される方法については、 構造化ストリーミングの本番運用に関する考慮事項を参照してください。

スキーマを明示的に更新して列を追加します

SQL

ALTER TABLE table_name ADD COLUMNS (col_name data_type [COMMENT col_comment] [FIRST|AFTER colA_name], ...)

デフォルトでは、NULL値の許容はtrueです。

ネストされたフィールドに列を追加するには、次のコマンドを使用します：

SQL

ALTER TABLE table_name ADD COLUMNS (col_name.nested_col_name data_type [COMMENT col_comment] [FIRST|AFTER colA_name], ...)

たとえば、ALTER TABLE boxes ADD COLUMNS (colB.nested STRING AFTER field1)を実行する前のスキーマが次のようになっているとします：

- root
| - colA
| - colB
| +-field1
| +-field2

その後のスキーマは次のとおりです：

- root
| - colA
| - colB
| +-field1
| +-nested
| +-field2

注記

ネストされた列の追加は、構造体に対してのみサポートされています。配列とマップはサポートされていません。

スキーマを明示的に更新して、列のコメントまたは順序を変更する

SQL

ALTER TABLE table_name ALTER [COLUMN] col_name (COMMENT col_comment | FIRST | AFTER colA_name)

ネストされたフィールドの列を変更するには、次のコマンドを使用します。

SQL

ALTER TABLE table_name ALTER [COLUMN] col_name.nested_col_name (COMMENT col_comment | FIRST | AFTER colA_name)

たとえば、ALTER TABLE boxes ALTER COLUMN colB.field2 FIRSTを実行する前のスキーマが次のようになっているとします：

- root
| - colA
| - colB
| +-field1
| +-field2

その後のスキーマは次のとおりです：

- root
| - colA
| - colB
| +-field2
| +-field1

スキーマを明示的に更新して列を置き換えます

SQL

ALTER TABLE table_name REPLACE COLUMNS (col_name1 col_type1 [COMMENT col_comment1], ...)

例えば、以下のようなDDLを実行する場合：

SQL

ALTER TABLE boxes REPLACE COLUMNS (colC STRING, colB STRUCT<field2:STRING, nested:STRING, field1:STRING>, colA STRING)

前のスキーマが次の場合：

- root
| - colA
| - colB
| +-field1
| +-field2

その後のスキーマは次のとおりです：

- root
| - colC
| - colB
| +-field2
| +-nested
| +-field1
| - colA

スキーマを明示的に更新して列の名前を変更する

注記

この機能は、Databricks Runtime 10.4 LTS 以降で使用できます。

列の既存のデータを書き換えずに列の名前を変更するには、テーブルの列マッピングを有効にする必要があります。「Delta Lake 列マッピングを使用した列の名前変更と削除」を参照してください。

列の名前を変更するには：

SQL

ALTER TABLE table_name RENAME COLUMN old_col_name TO new_col_name

ネストされたフィールドの名前を変更するには：

SQL

ALTER TABLE table_name RENAME COLUMN col_name.old_nested_field TO new_nested_field

たとえば、次のコマンドを実行するとします：

SQL

ALTER TABLE boxes RENAME COLUMN colB.field1 TO field001

以前のスキーマが次の場合：

- root
| - colA
| - colB
| +-field1
| +-field2

その後のスキーマは次のようになります：

- root
| - colA
| - colB
| +-field001
| +-field2

Delta Lake 列マッピングを使用した列の名前変更と削除を参照してください。

スキーマを明示的に更新して列をドロップする

注記

この機能は、Databricks Runtime 11.3 LTS 以降で使用できます。

データファイルを書き換えることなく、メタデータのみの操作として列をドロップするには、テーブルの列マッピングを有効にする必要があります。 Delta Lake 列マッピングを使用した列の名前変更と削除を参照してください。

important

メタデータから列を削除しても、ファイル内の列の基になるデータは削除されません。 ドロップされた列データをパージするには、 REORG TABLE を使用してファイルを書き換えます。 その後、vacuum を使用して、ドロップされた列データを含むファイルを物理的に削除できます。

列を削除するには：

SQL

ALTER TABLE table_name DROP COLUMN col_name

複数の列を削除するには：

SQL

ALTER TABLE table_name DROP COLUMNS (col_name_1, col_name_2)

スキーマを明示的に更新して列のタイプまたは名前を変更する

列のタイプや名前を変更したり、テーブルを書き換えて列を削除したりできます。これを行うには、 overwriteSchema オプションを使用します。

次の例は、列の型を変更する方法を示しています：

Python

(spark.read.table(...)
  .withColumn("birthDate", col("birthDate").cast("date"))
  .write
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .saveAsTable(...)
)

次の例は、列名の変更を示しています：

Python

(spark.read.table(...)
  .withColumnRenamed("dateOfBirth", "birthDate")
  .write
  .mode("overwrite")
  .option("overwriteSchema", "true")
  .saveAsTable(...)
)

スキーマ進化の有効化

スキーマ進化を有効にするには、次のいずれかを実行します。

• .option("mergeSchema", "true")を Spark データフレーム writeまたはwriteStream操作に設定します。新しい列を追加するには、書き込みのスキーマ進化の有効化を参照してください。
• MERGE WITH SCHEMA EVOLUTION構文を使用します。マージのスキーマ進化構文を参照してください。
• 現在の SparkSession の Spark conf spark.databricks.delta.schema.autoMerge.enabled を true に設定します。

Databricks は、 Spark conf を設定するのではなく、書き込み操作ごとにスキーマ進化を有効にすることをお勧めします。

オプションまたは構文を使用して書き込み操作でスキーマ進化を有効にする場合、これは Spark conf よりも優先されます。

注記

INSERT INTOステートメントのスキーマ進化句はありません。

新しい列を追加するための書き込みのスキーマ進化を有効にする

ソースクエリには存在するが、ターゲットテーブルには存在しないカラムは、スキーマ進化が有効な場合、書き込みトランザクションの一部として自動的に追加されます。 スキーマ進化の有効化を参照してください。

新しい列を追加するときに大文字と小文字が保持されます。 新しい列は、テーブル スキーマの最後に追加されます。 追加の列が構造体内にある場合、それらはターゲット テーブルの構造体の末尾に追加されます。

次の例は、 Auto Loaderで mergeSchema オプションを使用する方法を示しています。 「Auto Loaderとは」を参照してください。

Python

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", "<path-to-schema-location>")
  .load("<path-to-source-data>")
  .writeStream
  .option("mergeSchema", "true")
  .option("checkpointLocation", "<path-to-checkpoint>")
  .trigger(availableNow=True)
  .toTable("table_name")
)

次の例は、バッチ書き込み操作で mergeSchema オプションを使用する方法を示しています。

Python

(spark.read
  .table(source_table)
  .write
  .option("mergeSchema", "true")
  .mode("append")
  .saveAsTable("table_name")
)

Delta Lake マージの自動スキーマ進化

スキーマ進化を使用すると、マージでターゲットテーブルとソーステーブルの間のスキーマの不一致を解決できます。 次の 2 つのケースを処理します。

1. カラムはソース表に存在しますが、ターゲット表には存在しません。

   その列はターゲットスキーマに追加され、その値はソースの対応する列から入力されます。

   • これは、マージ ソースの列名と構造がターゲット割り当てと完全に一致する場合にのみ適用されます。

   • 新しい列は、式、名前変更、または変換を行わずに、ソースから直接割り当てる必要があります。

   これらの例では、スキーマ進化が可能になります。

   SQL

   UPDATE SET target.newcol = source.newcol  -- The column newcol will be added to target.
   UPDATE SET target.somestruct.newfield = source.somestruct.newfield  -- The field newfield will be added in struct column somestruct in target.
   UPDATE SET *  -- Any columns and nested fields in source that don't exist in target will be added to target.

   これらの例はスキーマ進化をトリガーしません。

   SQL

   UPDATE SET target.newcol = source.someothercol
   UPDATE SET target.newcol = source.x + source.y
   UPDATE SET target.newcol = source.output.newcol

2. カラムはターゲット表に存在しますが、ソース表には存在しません。

   ターゲット・スキーマは変更されません。これらの列は、次のとおりです。

   • UPDATE SET *は変更されません。

   • はINSERT *のNULLに設定されています。

   • action 句で割り当てられている場合は、明示的に変更される可能性があります。

   例えば：

   SQL

   UPDATE SET *  -- The target columns that are not in the source are left unchanged.
   INSERT *  -- The target columns that are not in the source are set to NULL.
   UPDATE SET target.onlyintarget = 5  -- The target column is explicitly updated.
   UPDATE SET target.onlyintarget = source.someothercol  -- The target column is explicitly updated from some other source column.

自動スキーマ進化を手動で有効にする必要があります。 スキーマ進化の有効化を参照してください。

注記

Databricks Runtime 12.2 LTS 以降では、ソース テーブルに存在する列と構造体フィールドを、挿入アクションまたは更新アクションで名前で指定できます。 Databricks Runtime 11.3 LTS 以前では、マージによるスキーマ進化に使用できるのは、INSERT * または UPDATE SET * アクションのみです。

Databricks Runtime 13.3 LTS 以降では、map<int, struct<a: int, b: int>> などのマップ内にネストされた構造体でスキーマ進化を使用できます。

マージのスキーマ進化構文

Databricks Runtime 15.4 LTS以降では、SQL または Delta テーブルを使用して merge ステートメントでスキーマ進化を指定できますAPIs

SQL

SQL

MERGE WITH SCHEMA EVOLUTION INTO target
USING source
ON source.key = target.key
WHEN MATCHED THEN
  UPDATE SET *
WHEN NOT MATCHED THEN
  INSERT *
WHEN NOT MATCHED BY SOURCE THEN
  DELETE

Python

Python

from delta.tables import *

(targetTable
  .merge(sourceDF, "source.key = target.key")
  .withSchemaEvolution()
  .whenMatchedUpdateAll()
  .whenNotMatchedInsertAll()
  .whenNotMatchedBySourceDelete()
  .execute()
)

Scala

Scala

import io.delta.tables._

targetTable
  .merge(sourceDF, "source.key = target.key")
  .withSchemaEvolution()
  .whenMatched()
  .updateAll()
  .whenNotMatched()
  .insertAll()
  .whenNotMatchedBySource()
  .delete()
  .execute()

スキーマ進化によるマージの操作例

ここでは、スキーマ進化を伴う場合と伴わない場合のmerge操作の影響の例をいくつか示します。

列	クエリー（SQL の場合）	スキーマ進化なしの動作（既定）	スキーマの進化に伴う動作
ターゲット列： key, value ソース列： key, value, new_value
ターゲット列： key, old_value ソース列： key, new_value
ターゲット列： key, old_value ソース列： key, new_value
ターゲット列： key, old_value ソース列： key, new_value

(1) この動作は、Databricks Runtime 12.2 LTS 以降で使用できます。Databricks Runtime 11.3 LTS 以下では、この条件でエラーが発生します。

Delta Lake マージで列を除外する

Databricks Runtime 12.2 LTS 以降では、マージ条件で EXCEPT 句を使用して列を明示的に除外できます。 EXCEPT キーワードの動作は、スキーマ進化が有効になっているかどうかによって異なります。

スキーマ進化が無効な場合、EXCEPT キーワードがターゲットテーブルの列のリストに適用され、UPDATE または INSERT アクションから列を除外できるようになります。除外された列は null に設定されます。

スキーマ進化が有効な場合、EXCEPT キーワードがソーステーブルの列のリストに適用され、スキーマ進化から列を除外することができます。ターゲットに存在しないソースの新しい列が EXCEPT 句にリストされている場合、その列はターゲットスキーマに追加されません。除外される列がターゲットに既に存在する場合、null に設定されます。

次の例は、構文を示しています。

列	クエリー（SQL の場合）	スキーマ進化なしの動作（既定）	スキーマの進化に伴う動作
ターゲット列： id, title, last_updated ソース列： id, title, review, last_updated
ターゲット列： id, title, last_updated ソース列： id, title, review, internal_count

スキーマ更新の NullType 列の扱い

Parquet は NullTypeをサポートしていないため、Delta テーブルに書き込むときに NullType 列は DataFrame から削除されますが、スキーマには引き続き格納されます。その列に対して異なるデータ型が受信されると、Delta Lake はスキーマを新しいデータ型にマージします。Delta Lake が既存の列の NullType を受け取った場合、古いスキーマは保持され、新しい列は書き込み中に削除されます。

NullType ストリーミングではサポートされていません。ストリーミングを使用する場合はスキーマを設定する必要があるため、これは非常にまれです。NullTypeは、ArrayTypeやMapTypeなどの複合型にも受け入れられません。

テーブル スキーマの置換

デフォルトでは、テーブル内のデータを上書きしてもスキーマは上書きされません。replaceWhereを使用せずにmode("overwrite")を使用してテーブルを上書きする場合でも、書き込まれるデータのスキーマを上書きする必要がある場合があります。overwriteSchemaオプションをtrueに設定して、テーブルのスキーマとパーティションを置き換えます：

Python

df.write.option("overwriteSchema", "true")

important

動的パーティションの上書きを使用する場合、overwriteSchemaをtrueとして指定することはできません。