Выборочно перезаписывать данные с помощью Delta Lake

Delta Lake имеет следующие варианты отличительного выборочного перезаписывания:

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

Для большинства операций Databricks рекомендует указать replaceWhere , какие данные следует перезаписать.

Important

Если данные были случайно перезаписаны, можно использовать восстановление для отмены изменения.

Произвольная выборочная перезапись с помощью replaceWhere

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

Note

ДЛЯ SQL требуется Databricks Runtime 12.2 LTS или более поздней версии.

Например, атомарно замените события в январе в целевой таблице, которая секционируется по start_date, данными из 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

Note

REPLACE WHERE принимает boolean_expression с некоторыми ограничениями. Смотрите INSERT в справочнике по языку SQL.

Устаревшее поведение

При использовании устаревшего replaceWhereповедения запросы перезаписывают данные, соответствующие предикату только по столбцам секций. Следующая команда будет атомарно заменять месяц январь в целевой таблице, которая секционирована с помощью date, данными из 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

Динамические перезаписи разделов

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

Azure Databricks рекомендует REPLACE USING для активации динамических перезаписей секций. Этот режим работает во всех типах вычислений, включая хранилища Databricks SQL, бессерверные вычисления и классические вычисления, и не требует настройки сеанса Spark. См. перезаписи динамических разделов с REPLACE USING.

partitionOverwriteMode — это устаревший режим для динамических перезаписей секций, которые требуют использования классических вычислений и настройки сеанса Spark. Он не поддерживается в databricks SQL или бессерверных вычислениях. См. перезаписи динамических разделов с помощью partitionOverwriteMode (устаревшая версия).

В следующих разделах показано, как использовать каждый режим.

Динамические перезаписи разделов с помощью REPLACE USING

Databricks Runtime 16.3 и более поздних версий поддерживает динамические перезаписи секций для таблиц с помощью REPLACE USING. Вы можете выборочно перезаписать данные во всех типах вычислений, не устанавливая конфигурацию сеанса Spark. REPLACE USING обеспечивает независимое от вычислений, атомарное поведение перезаписи, которое работает на хранилищах SQL Databricks, бессерверных вычислениях и классических вычислениях.

REPLACE USING перезаписывает секции, предназначенные для входящих данных. Все остальные разделы остаются неизменными.

REPLACE USING поддерживается только в SQL. Дополнительные сведения см. в INSERT справочнике по языку SQL.

В следующих примерах используется REPLACE USING:

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

Помните о следующих ограничениях и особенностях динамической перезаписи разделов:

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

Если вам нужна более настраиваемая логика сопоставления, чем то, что поддерживает REPLACE USING, например, рассматривать значения NULL как равные, используйте комплементарное REPLACE ON. Подробные сведения см. в разделе INSERT.

Динамические перезаписи секций с partitionOverwriteMode (устаревшей версией)

Important

Эта функция доступна в общедоступной предварительной версии.

Databricks Runtime, начиная с версии 11.3 LTS, поддерживает динамическое переписывание секционированных таблиц в режиме перезаписи: INSERT OVERWRITE в SQL или запись в DataFrame с df.write.mode("overwrite"). Этот тип перезаписи доступен только для классических вычислений, а не для хранилищ SQL Databricks или бессерверных вычислений.

Предупреждение

По возможности используйте INSERT OVERWRITE REPLACE USING вместо перезаписи разделов INSERT OVERWRITE PARTITION и spark.sql.sources.partitionOverwriteMode=dynamic. Перезапись раздела может использовать устаревшие данные при изменении секционирования.

Чтобы использовать режим динамической перезаписи секции, задайте для конфигурации сеанса Spark значение spark.sql.sources.partitionOverwriteModedynamic. Кроме того, можно задать опцию DataFrameWriterpartitionOverwriteMode в 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 .
  • Всегда проверяйте, что записанные данные касаются только ожидаемых секций. Одна строка в неправильной секции может непреднамеренно перезаписать всю секцию.
  • Last updated on