Обновление схемы таблицы

Таблицы поддерживают эволюцию схемы, что позволяет изменять структуру таблицы по мере изменения требований к данным. Поддерживаются следующие типы изменений.

• Добавление новых столбцов при произвольных позициях
• Изменение порядка существующих столбцов
• Переименование существующих столбцов
• Изменение типов существующих столбцов, см. раздел "Изменение ширины типов с автоматической эволюцией схемы"

Внесите эти изменения явным образом с помощью DDL или неявно с помощью DML.

Внимание

Обновления схемы конфликтуют со всеми параллельными операциями записи. Databricks рекомендует координировать изменения схемы во избежание конфликтов записи.

Обновление схемы таблицы завершает любые потоки, считываемые из этой таблицы. Чтобы продолжить обработку, перезапустите поток с помощью методов, описанных в рекомендациях по рабочей среде для структурированной потоковой передачи.

Явное обновление схемы для добавления столбцов

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

По умолчанию допустимость значений NULL установлена как true.

Чтобы добавить столбец во вложенное поле, используйте:

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

Примечание.

Добавление вложенных столбцов поддерживается только для структур. Массивы и карты не поддерживаются.

Явно обновите схему, чтобы изменить комментарии столбцов или их порядок.

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

Чтобы изменить столбец во вложенном поле, используйте:

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

Явное изменение схемы для замены столбцов

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

Например, при выполнении следующего DDL:

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.

Переименование столбцов:

ALTER TABLE table_name RENAME COLUMN old_col_name TO new_col_name

Чтобы переименовать вложенное поле, сделайте следующее:

ALTER TABLE table_name RENAME COLUMN col_name.old_nested_field TO new_nested_field

Например, при выполнении следующей команды:

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.

Внимание

Удаление столбца из метаданных не приводит к фактическому удалению данных этих столбцов из файлов данных. Чтобы очистить удаленные данные столбца, можно использовать REORG TABLE для перезаписи файлов. Затем можно использовать VACUUM для физического удаления файлов, содержащих данные удаленного столбца.

Чтобы удалить столбец, выполните приведенные действия.

ALTER TABLE table_name DROP COLUMN col_name

Чтобы удалить несколько столбцов, выполните следующее.

ALTER TABLE table_name DROP COLUMNS (col_name_1, col_name_2)

Выполните явное обновление схемы для изменения типа столбца или имени

Вы можете изменить тип или имя столбца или удалить столбец, перезаписав таблицу. Для этого используйте параметр overwriteSchema.

В следующем примере показано изменение типа столбца:

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

В следующем примере показано изменение имени столбца:

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

Включение эволюции схемы

Включите эволюцию схемы с помощью одного из следующих методов:

• Использование INSERT WITH SCHEMA EVOLUTION синтаксиса: Работает с операторами INSERT. Включите WITH SCHEMA EVOLUTION в синтаксис SQL.
• Использование MERGE WITH SCHEMA EVOLUTION синтаксиса: Работает с операторами MERGE. Включите WITH SCHEMA EVOLUTION в синтаксис SQL или используйте .withSchemaEvolution() в API Azure Databricks.
• mergeSchema Задайте параметр: работает с пакетной записью или потоковой записью. Задайте .option("mergeSchema", "true") для отдельных операций записи.
• Задайте конфигурацию Spark (устаревшая версия): Задает spark.databricks.delta.schema.autoMerge.enabled значение true для всего SparkSession. Не рекомендуется для использования в производственной среде.

Databricks рекомендует включать эволюцию схемы для каждой операции записи с помощью синтаксиса WITH SCHEMA EVOLUTION или опции mergeSchema, а не настройки конфигурации Spark.

При использовании параметров или синтаксиса для включения эволюции схемы в операции записи это имеет приоритет над конфигурацией Spark.

Включение эволюции схемы для добавления новых столбцов при записи

Столбцы, которые присутствуют в исходном запросе, но отсутствующие из целевой таблицы, автоматически добавляются в рамках транзакции записи при включении эволюции схемы. См. раздел "Включить эволюцию схемы".

Регистр сохраняется при добавлении нового столбца. Новые столбцы добавляются в конец схемы таблицы. Если дополнительные столбцы находятся в структуре, они добавляются в конец структуры в целевой таблице.

Синтаксис эволюции схемы для INSERT

Вы можете использовать WITH SCHEMA EVOLUTION в инструкциях INSERT для включения эволюции схемы.

INSERT WITH SCHEMA EVOLUTION INTO target_table
SELECT * FROM source_table

Если запрос на source_table возвращает столбцы, которые не существуют в целевой таблице, эти столбцы автоматически добавляются в схему target_table. Существующие строки получают NULL значения для новых столбцов.

INSERT с эволюцией схемы с помощью API кадра данных

В следующем примере показано использование mergeSchema параметра с пакетной операцией записи:

Python

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

язык программирования Scala

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

INSERT с эволюцией схемы в потоковой передаче

В следующем примере показано использование опции mergeSchema с Auto Loader. См. статью об автозагрузчике.

(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")
)

Автоматическая эволюция схемы для слияния

Эволюция схемы позволяет разрешать несоответствия схемы между целевой и исходной таблицей в слиянии. Он обрабатывает следующие два случая:

1. Столбец существует в исходной таблице, но не целевой таблице, и указывается по имени в назначении действий вставки или обновления. Либо присутствует UPDATE SET *, либо INSERT * действие.

   Этот столбец будет добавлен в целевую схему, а его значения будут заполнены из соответствующего столбца в источнике.

   • Это применяется только в том случае, если имя столбца и структура в источнике слияния точно соответствуют целевому назначению.

   • Новый столбец должен присутствовать в исходной схеме. Назначение нового столбца в предложении действия не определяет этот столбец.

   Эти примеры позволяют эволюции схемы:

   -- The column newcol is present in the source but not in the target. It will be added to the target.
   UPDATE SET target.newcol = source.newcol
   
   -- The field newfield doesn't exist in struct column somestruct of the target. It will be added to that struct column.
   UPDATE SET target.somestruct.newfield = source.somestruct.newfield
   
   -- The column newcol is present in the source but not in the target.
   -- It will be added to the target.
   UPDATE SET target.newcol = source.newcol + 1
   
   -- Any columns and nested fields in the source that don't exist in target will be added to the target.
   UPDATE SET *
   INSERT *
   

   Эти примеры не активируют эволюцию схемы, если столбец newcol отсутствует в схеме source :

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

2. Столбец существует в целевой таблице, но не в исходной таблице.

   Целевая схема не изменяется. Эти столбцы:

   • Остаются неизменными для UPDATE SET *.

   • Задано значение NULLINSERT *.

   • Изменение все еще может быть явным, если оно назначено в оговорке о действиях.

   Рассмотрим пример.

   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 и более поздних версиях можно указать эволюцию схемы в инструкции слияния с помощью API 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

from delta.tables import *

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

язык программирования 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	MERGE INTO target_table t USING source_table s ON t.key = s.key WHEN MATCHED THEN UPDATE SET * WHEN NOT MATCHED THEN INSERT *	Схема таблицы остается неизменной; обновляются или вставляются только столбцы key и value.	Схема таблицы меняется на (key, value, new_value). Существующие записи с совпадениями обновляются value и new_value в источнике. Новые строки вставляются со схемой (key, value, new_value).
Целевые столбцы: key, old_value Исходные столбцы: key, new_value	MERGE INTO target_table t USING source_table s ON t.key = s.key WHEN MATCHED THEN UPDATE SET * WHEN NOT MATCHED THEN INSERT *	Действия UPDATE и INSERT вызывают ошибку, так как целевой столбец old_value не находится в источнике.	Схема таблицы меняется на (key, old_value, new_value). Существующие записи с совпадениями обновляются в источнике new_value, оставляя old_value без изменений. Новые записи вставляются с указанными key, new_value и NULL для old_value.
Целевые столбцы: key, old_value Исходные столбцы: key, new_value	MERGE INTO target_table t USING source_table s ON t.key = s.key WHEN MATCHED THEN UPDATE SET new_value = s.new_value	UPDATE выдает ошибку, поскольку столбец new_value не существует в целевой таблице.	Схема таблицы меняется на (key, old_value, new_value). Существующие записи с совпадениями обновляются в new_value источнике, оставляя old_value без изменений, а для несоответствующих записей введено NULL для new_value. См. примечание (1).
Целевые столбцы: key, old_value Исходные столбцы: key, new_value	MERGE INTO target_table t USING source_table s ON t.key = s.key WHEN NOT MATCHED THEN INSERT (key, new_value) VALUES (s.key, s.new_value)	INSERT выдает ошибку, поскольку столбец new_value не существует в целевой таблице.	Схема таблицы меняется на (key, old_value, new_value). Новые записи вставляются с указанными key, new_value и NULL для old_value. Существующие записи введены NULL для new_value, оставляя old_value без изменений. См. примечание (1).

(1) Это поведение доступно в Databricks Runtime 12.2 LTS и выше; в Databricks Runtime 11.3 LTS и ниже возникает ошибка в этом условии.

Исключение столбцов с объединением

В Databricks Runtime 12.2 LTS и более поздних версиях можно использовать EXCEPT предложения в условиях слияния для явного исключения столбцов. Поведение ключевого EXCEPT слова зависит от того, включена эволюция схемы или нет.

При отключенной EXCEPT эволюции схемы ключевое слово применяется к списку столбцов в целевой таблице и позволяет исключить столбцы из UPDATE или INSERT действий. Для исключенных столбцов задано значение null.

При включенной EXCEPT эволюции схемы ключевое слово применяется к списку столбцов в исходной таблице и позволяет исключить столбцы из эволюции схемы. Новый столбец в источнике, который отсутствует в целевом объекте, не добавляется в целевую схему, если он указан в предложении EXCEPT . Исключенные столбцы, которые уже присутствуют в целевом объекте, имеют nullзначение .

В следующих примерах показан этот синтаксис:

Столбцы	Запрос (в SQL)	Поведение без развития схемы (по умолчанию)	Поведение при развитии схемы
Целевые столбцы: id, title, last_updated Исходные столбцы: id, title, review, last_updated	MERGE INTO target t USING source s ON t.id = s.id WHEN MATCHED THEN UPDATE SET last_updated = current_date() WHEN NOT MATCHED THEN INSERT * EXCEPT (last_updated)	Сопоставленные строки обновляются путем задания last_updated поля текущей дате. Новые строки вставляются с помощью значений для id и title. Для исключенного поля задано значение last_updatednull. Поле review игнорируется, так как оно не находится в целевом объекте.	Сопоставленные строки обновляются путем задания last_updated поля текущей дате. Схема обновляется для добавления поля review. Новые строки вставляются с помощью всех исходных полей, за исключением last_updated, которому присваивается null.
Целевые столбцы: id, title, last_updated Исходные столбцы: id, title, review, internal_count	MERGE INTO target t USING source s ON t.id = s.id WHEN MATCHED THEN UPDATE SET last_updated = current_date() WHEN NOT MATCHED THEN INSERT * EXCEPT (last_updated, internal_count)	INSERT выдает ошибку, поскольку столбец internal_count не существует в целевой таблице.	Сопоставленные строки обновляются путем задания last_updated поля текущей дате. Поле review добавляется в целевую таблицу, но internal_count поле игнорируется. Для новых вставленных строк задано last_updated значение null.

Включение эволюции схемы с помощью конфигурации Spark (устаревшая версия)

Конфигурацию Spark можно настроить spark.databricks.delta.schema.autoMerge.enabled для включения эволюции схемы true для всех операций записи в текущем SparkSession.

Python

spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", True)

язык программирования Scala

spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", true)

SQL

SET spark.databricks.delta.schema.autoMerge.enabled=true

Внимание

Этот подход не рекомендуется для использования в рабочей среде. Вместо этого включите эволюцию схемы для каждой операции записи:

• Для INSERT и пакетной или потоковой записи, используйте .option("mergeSchema", "true") или INSERT WITH SCHEMA EVOLUTION
• Для MERGE инструкций используйте MERGE WITH SCHEMA EVOLUTION

Настройка конфигурации на уровне сеанса может привести к непреднамеренным изменениям схемы в нескольких операциях и усложнить понимание того, какие операции изменяют схему.

При использовании параметров или синтаксиса для включения эволюции схемы в операции записи это имеет приоритет над конфигурацией Spark.

Замена схемы таблицы

По умолчанию перезапись данных в таблице не приводит к перезаписи схемы. При перезаписи таблицы, использующей mode("overwrite") без replaceWhere, вы по-прежнему можете перезаписать схему записываемых данных. Замените схему и секционирование таблицы, задав для параметра overwriteSchema значение true:

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

Внимание

Невозможно указать overwriteSchema , как true при использовании динамической перезаписи секции.