Расширение типов

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

Для расширения типов данных требуется Delta Lake. Все управляемые таблицы каталога Unity используют Delta Lake по умолчанию.

Поддерживаемые изменения типов

Типы можно расширить в соответствии со следующими правилами:

Чтобы избежать случайного преобразования целочисленных значений в десятичные, необходимо вручную зафиксировать изменения типов из byte, short, int, или long на decimal или double. При приведении целочисленного типа к decimal или double, если какой-либо поток записывает это значение обратно в целочисленный столбец, Spark усечет дробную часть значений по умолчанию.

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

Включение расширения типов

Вы можете включить расширение типов в существующей таблице, задав для свойства таблицы delta.enableTypeWidening значение true:

  ALTER TABLE <table_name> SET TBLPROPERTIES ('delta.enableTypeWidening' = 'true')

Вы также можете включить расширение типов во время создания таблицы:

  CREATE TABLE T(c1 INT) TBLPROPERTIES('delta.enableTypeWidening' = 'true')

Применение изменения типа вручную

ALTER COLUMN Используйте команду для ручного изменения типов:

ALTER TABLE <table_name> ALTER COLUMN <col_name> TYPE <new_type>

Эта операция обновляет схему таблицы без перезаписи базовых файлов данных.

Расширенные типы с автоматической эволюцией схемы

Эволюция схемы работает с расширением типов для обновления типов данных в целевых таблицах в соответствии с типом входящих данных.

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

• Команда записи выполняется с включенной автоматической эволюцией схемы.
• Целевая таблица имеет поддержку расширения типа.
• Тип исходного столбца шире, чем тип целевого столбца.
• Процесс расширения типов поддерживает изменение типа данных.
• Изменение типа не является одним из byte, short, int, long или decimal или double. Эти изменения типов можно применять только вручную с помощью ALTER TABLE, чтобы избежать случайного преобразования целых чисел в десятичные типы.

Несоответствия типов, которые не соответствуют всем этим условиям, соответствуют обычным правилам применения схемы. См. соблюдение схемы.

Отключение функции расширения типов данных в таблице

Чтобы предотвратить расширение случайного типа в включенных таблицах, присвоив свойству значение false:

  ALTER TABLE <table_name> SET TBLPROPERTIES ('delta.enableTypeWidening' = 'false')

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

Если необходимо полностью удалить функции таблицы расширения типа, можно использовать команду DROP FEATURE, как показано в следующем примере:

 ALTER TABLE <table-name> DROP FEATURE 'typeWidening' [TRUNCATE HISTORY]

При удалении расширения типов все файлы данных, которые не соответствуют текущей схеме таблицы, перезаписываются. См. Управление функцией удаления таблицы Delta Lake и понижение протокола таблицы.

Потоковая передача из таблицы Delta

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

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

Для каждого потока чтения для источника данных должен быть указан свой schemaTrackingLocation. Указанный параметр schemaTrackingLocation должен содержаться в каталоге, указанном для checkpointLocation целевой таблицы для потоковой записи.

Параметр schemaTrackingLocation используется для указания пути для отслеживания схем, как показано в следующем примере кода:

Питон

checkpoint_path = "/path/to/checkpointLocation"

(spark.readStream
  .option("schemaTrackingLocation", checkpoint_path)
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .toTable("output_table")
)

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

val checkpointPath = "/path/to/checkpointLocation"

spark.readStream
  .option("schemaTrackingLocation", checkpointPath)
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", checkpointPath)
  .toTable("output_table")

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

Чтобы возобновить обработку, задайте конфигурацию spark.databricks.delta.streaming.allowSourceColumnTypeChange Spark или параметр allowSourceColumnTypeChange средства чтения DataFrame.

Питон

checkpoint_path = "/path/to/checkpointLocation"

(spark.readStream
  .option("schemaTrackingLocation", checkpoint_path)
  .option("allowSourceColumnTypeChange", "<delta_source_table_version>")
  # alternatively to allow all future type changes for this stream:
  # .option("allowSourceColumnTypeChange", "always")
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .toTable("output_table")
)

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

val checkpointPath = "/path/to/checkpointLocation"

spark.readStream
  .option("schemaTrackingLocation", checkpointPath)
  .option("allowSourceColumnTypeChange", "<delta_source_table_version>")
  // alternatively to allow all future type changes for this stream:
  // .option("allowSourceColumnTypeChange", "always")
  .table("delta_source_table")
  .writeStream
  .option("checkpointLocation", checkpointPath)
  .toTable("output_table")

SQL

  -- To unblock for this particular stream just for this series of schema change(s):
  SET spark.databricks.delta.streaming.allowSourceColumnTypeChange.ckpt_<checkpoint_id> = "<delta_source_table_version>"
  -- To unblock for this particular stream:
  SET spark.databricks.delta.streaming.allowSourceColumnTypeChange = "<delta_source_table_version>"
  -- To unblock for all streams:
  SET spark.databricks.delta.streaming.allowSourceColumnTypeChange = "always"

Идентификатор <checkpoint_id> контрольной точки и версия <delta_source_table_version> исходной таблицы Delta Lake отображаются в сообщении об ошибке при остановке потока.

Delta Sharing (Дельта шаринг)

Совместное использование таблицы Delta Lake с включенным расширением типов поддерживается в Databricks to Databricks Delta Sharing. Поставщик и получатель должны находиться в Databricks Runtime 16.1 или более поздней версии.

Чтобы прочитать поток данных изменений из таблицы Delta Lake с включенным расширением типа с помощью Delta Sharing, необходимо задать формат ответа следующим образом: delta.

spark.read
  .format("deltaSharing")
  .option("responseFormat", "delta")
  .option("readChangeFeed", "true")
  .option("startingVersion", "<start version>")
  .option("endingVersion", "<end version>")
  .load("<table>")

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

Ограничения

Совместимость Айсберг

Айсберг не поддерживает все изменения типа, охватываемые расширением типа, см. в разделе "Эволюция схемы Айсберга". В частности, Azure Databricks не поддерживает следующие изменения типа:

• byte, shortintlong или decimaldouble
• увеличение десятичного масштабирования
• date до timestampNTZ

Если в таблице Delta Lake включена совместимость UniForm с Iceberg, применение одной из этих изменений типов приводит к ошибке.

При применении одного из этих неподдерживаемых изменений типа к таблице Delta Lake включение совместимости с Iceberg в таблице приводит к ошибке. Чтобы устранить эту ошибку, необходимо удалить функцию расширения таблицы типа.

Другие ограничения

• Предоставление местоположения отслеживания схемы с помощью SQL при потоковой передаче из таблицы Delta Lake, в которой произошли изменения типа, не поддерживается.
• Совместное использование таблицы, используя Delta Sharing с включенной функцией расширения типов, с потребителями, не являющимися пользователями Databricks, не поддерживается.