Обновление схем таблиц с помощью эволюции схемы

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

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

Внимание

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

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

Ручные изменения схемы

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

Добавление столбцов

Используйте ALTER TABLE ... ADD COLUMNS для добавления одного или нескольких столбцов в существующую таблицу, при необходимости указывая позицию и комментарий:

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 ... ALTER COLUMN для обновления комментария или переупорядочения столбца относительно других столбцов:

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 ... REPLACE COLUMNS для переопределения полного списка столбцов таблицы, включая добавление, удаление, изменение порядка столбцов или переименование столбцов в одной операции:

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

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

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

Удалить столбцы

Чтобы удалить столбцы, оперируя только с метаданными, то есть без перезаписи файлов данных, для этой таблицы должно быть включено сопоставление столбцов. См. как переименовывать и удалять столбцы с использованием сопоставления столбцов в 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(...)
)

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

Используйте WITH SCHEMA EVOLUTION или установите mergeSchema в значение true, чтобы внести изменения в схему на основе схемы данных, которые требуется INSERT или MERGE в существующую таблицу.

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

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

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

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

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

Рассмотрим следующее:

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

INSERT с эволюцией схемы с помощью SQL

В Databricks Runtime 18.1 и более поздних версиях используйте конструкцию WITH SCHEMA EVOLUTION в операторах INSERT, чтобы включить поддержку эволюции схемы:

INSERT WITH SCHEMA EVOLUTION INTO target_table
SELECT * FROM source_table

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

Конструкция WITH SCHEMA EVOLUTION поддерживает формы INSERT INTO, INSERT OVERWRITE и INSERT INTO ... REPLACE. Целевым объектом должна быть таблица Delta Lake или Apache Iceberg. Вставка в таблицу Hive или любую другую таблицу, кроме Delta, с этим условием приводит к ошибке.

Если вы используете Databricks Runtime версии 18.0 и ниже, включите эволюцию схемы с помощью параметра mergeSchema вместо этого. См. INSERT эволюцию схемы с использованием API DataFrame.

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 с эволюцией схемы в Structured Streaming

В следующем примере показано, как использовать параметр mergeSchema с Auto Loader для Structured Streaming. См. раздел "Что такое автозагрузчик?".

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

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

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

  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 11.3 LTS и ниже можно использовать только действия INSERT * или UPDATE SET * для эволюции схемы при слиянии.

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

В Databricks Runtime 13.3 LTS и более поздних версиях можно использовать эволюцию схемы со структурами, вложенными в отображения, например map<int, struct<a: int, b: int>>.

MERGEс эволюцией схемы с помощью SQL, Python и Scala

В 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 с эволюцией схемы

Ниже приведено несколько примеров влияния операции MERGE с развитием схемы и без него.

Columns Запрос (в 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значение .

Примеры EXCLUDE с MERGE

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

Columns Запрос (в 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

Примечание.

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

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

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

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

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

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

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

Примечание.

Невозможно указать overwriteSchema , как true при использовании динамической перезаписи секции. См. перезаписи динамических разделов с помощью partitionOverwriteMode (устаревшая версия).

  • Last updated on