Работа с историей таблицы

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

Получение истории таблицы

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

Хранение журнала таблиц определяется параметром logRetentionDurationтаблицы, который составляет 30 дней по умолчанию.

DESCRIBE HISTORY table_name       -- get the full history of the table

DESCRIBE HISTORY table_name LIMIT 1  -- get the last operation only

Сведения о синтаксисе Spark SQL см. в DESCRIBE HISTORY.

Дополнительные сведения о синтаксисе Scala/Java/Python см. в документации по API Delta Lake.

Обозреватель каталогов предоставляет визуальное представление этой подробной информации о таблице и её истории. Помимо схемы таблицы и выборки данных, можно также щелкнуть вкладку История, чтобы просмотреть журнал таблицы, который отображается с DESCRIBE HISTORY.

Схема истории

Выходные данные операции history имеют указанніе ниже столбцы.

+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+
|version|          timestamp|userId|userName|operation| operationParameters| job|notebook|clusterId|readVersion|   isolationLevel|isBlindAppend|    operationMetrics|
+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+
|      5|2019-07-29 14:07:47|   ###|     ###|   DELETE|[predicate -> ["(...|null|     ###|      ###|          4|WriteSerializable|        false|[numTotalRows -> ...|
|      4|2019-07-29 14:07:41|   ###|     ###|   UPDATE|[predicate -> (id...|null|     ###|      ###|          3|WriteSerializable|        false|[numTotalRows -> ...|
|      3|2019-07-29 14:07:29|   ###|     ###|   DELETE|[predicate -> ["(...|null|     ###|      ###|          2|WriteSerializable|        false|[numTotalRows -> ...|
|      2|2019-07-29 14:06:56|   ###|     ###|   UPDATE|[predicate -> (id...|null|     ###|      ###|          1|WriteSerializable|        false|[numTotalRows -> ...|
|      1|2019-07-29 14:04:31|   ###|     ###|   DELETE|[predicate -> ["(...|null|     ###|      ###|          0|WriteSerializable|        false|[numTotalRows -> ...|
|      0|2019-07-29 14:01:40|   ###|     ###|    WRITE|[mode -> ErrorIfE...|null|     ###|      ###|       null|WriteSerializable|         true|[numFiles -> 2, n...|
+-------+-------------------+------+--------+---------+--------------------+----+--------+---------+-----------+-----------------+-------------+--------------------+

Общие сведения partitionBy о параметрах операции

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

Для операций добавления к существующим таблицам (APPEND, INSERT, UPDATEDELETE, MERGE) это поле может отображать пустой массив [] или столбцы секций в зависимости от используемого метода записи (.save() vs .saveAsTable()). Эта несогласованность является ожидаемым состоянием и не должна использоваться для проверки записей.

Example

Рассмотрим таблицу, секционированную по столбцу date :

# Initial table creation - partitionBy is populated
df.write.format("delta") \
  .partitionBy("date") \
  .saveAsTable("sales_data")

Операция CREATE в журнале показывает:

operationParameters: {
  "mode": "ErrorIfExists",
  "partitionBy": "[\"date\"]"
}

При добавлении данных в эту таблицу:

# Subsequent append - partitionBy shows empty
new_df.write.format("delta") \
  .mode("append") \
  .saveAsTable("sales_data")

Операция APPEND показывает:

operationParameters: {
  "mode": "Append",
  "partitionBy": "[]"
}

Ожидается пустое partitionBy значение. Данные по-прежнему записываются в правильные секции на основе существующей схемы секционирования таблицы. Обратите внимание, что .save() в пути могут отображаться столбцы секций в этом поле, но это различие является подробной реализацией и не влияет на поведение записи.

Метрики операций

Операция history возвращает коллекцию метрик операций в карте столбцов operationMetrics.

В следующих таблицах перечислены определения ключей карты по операциям.

Что такое путешествие по времени?

Путешествие во времени поддерживает выполнение запросов к предыдущим версиям таблиц на основе временной метки или версии таблицы (как это записано в журнале транзакций). Для таких приложений можно использовать поездку по времени:

• Воссоздание анализа, отчетов или выходных данных (например, выходных данных модели машинного обучения). Это может быть полезно для отладки или аудита, особенно в регулируемых отраслях.
• Написание сложных темпоральных запросов.
• Устранение ошибок в данных.
• Обеспечение изоляции моментальных снимков для набора запросов при работе с таблицами с частыми изменениями.

Синтаксис перемещения по времени

Чтобы запросить таблицу со временем, добавьте предложение после спецификации имени таблицы.

• timestamp_expression может быть одним из следующих вариантов:
  • '2018-10-18T22:15:12.013Z', то есть строка, которая может приводиться к метке времени
  • cast('2018-10-18 13:36:32 CEST' as timestamp)
  • '2018-10-18', то есть строка даты.
  • current_timestamp() - interval 12 hours
  • date_sub(current_date(), 1)
  • Любое другое выражение, которое является меткой времени или может быть приведено к ней
• version — это длинное значение, которое можно получить из выходных данных DESCRIBE HISTORY table_spec.

Ни timestamp_expression, ни version не может быть подзапросом.

Принимаются только строки метки даты или времени. Например, "2019-01-01" и "2019-01-01T00:00:00.000Z". См. следующий код для примера синтаксиса:

SQL

SELECT * FROM people10m TIMESTAMP AS OF '2018-10-18T22:15:12.013Z';
SELECT * FROM people10m VERSION AS OF 123;

Питон

df1 = spark.read.option("timestampAsOf", "2019-01-01").table("people10m")
df2 = spark.read.option("versionAsOf", 123).table("people10m")

Можно также использовать @ синтаксис, чтобы указать метку времени или версию в составе имени таблицы. Метка времени должна быть указана в формате yyyyMMddHHmmssSSS. Вы можете указать версию после @, добавив к версии v. Посмотрите следующий код, чтобы ознакомиться с примером синтаксиса:

SQL

SELECT * FROM people10m@20190101000000000
SELECT * FROM people10m@v123

Питон

spark.read.table("people10m@20190101000000000")
spark.read.table("people10m@v123")

Что такое контрольные точки журнала транзакций?

Версии таблиц записываются в виде JSON-файлов в каталоге журнала транзакций, который хранится вместе с данными таблицы. Для оптимизации запросов к проверочным точкам версии таблиц объединяются в файлы контрольных точек Parquet, что предотвращает необходимость считывания всех версий истории таблиц в формате JSON. Azure Databricks оптимизирует частоту контрольных точек для размера данных и рабочей нагрузки. Пользователям не нужно взаимодействовать с контрольными точками напрямую. Частота контрольных точек подлежит изменению без уведомления.

Настройка хранения данных для запросов на поездки по времени

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

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

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

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

• delta.logRetentionDuration = "interval <interval>": управляет продолжительностью хранения истории для таблицы. Значение по умолчанию — interval 30 days.
• delta.deletedFileRetentionDuration = "interval <interval>": определяет пороговое значение VACUUM , используемое для удаления файлов данных, на которые больше не ссылается текущая версия таблицы. Значение по умолчанию — interval 7 days.

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

Восстановление таблицы в более раннее состояние

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

RESTORE TABLE target_table TO VERSION AS OF <version>;
RESTORE TABLE target_table TO TIMESTAMP AS OF <timestamp>;

Сведения о синтаксисе см. в RESTORE.

Восстановить метрики

RESTORE передает следующие метрики в виде DataFrame из одной строки после завершения операции:

• table_size_after_restore — размер таблицы после восстановления;

• num_of_files_after_restore — число файлов в таблице после восстановления;

• num_removed_files — число файлов, удаленных (логически удаленных) из таблицы;

• num_restored_files — число файлов, восстановленных из-за отката;

• removed_files_size — общий размер в байтах файлов, удаленных из таблицы;

• restored_files_size — общий размер восстановленных файлов в байтах.

  

Примеры использования путешествия по времени

• Исправление случайных удалений в таблице для пользователя 111:

  INSERT INTO my_table
    SELECT * FROM my_table TIMESTAMP AS OF date_sub(current_date(), 1)
    WHERE userId = 111
  

• Исправление случайного неправильного обновления таблицы:

  MERGE INTO my_table target
    USING my_table TIMESTAMP AS OF date_sub(current_date(), 1) source
    ON source.userId = target.userId
    WHEN MATCHED THEN UPDATE SET *
  

• Запросите количество новых клиентов, добавленных за последнюю неделю.

  SELECT
  (
    SELECT count(distinct userId)
    FROM my_table
  )
  -
  (
    SELECT count(distinct userId)
    FROM my_table TIMESTAMP AS OF date_sub(current_date(), 7)
  ) AS new_customers
  

Как найти номер версии последнего коммита в сеансе Spark?

Чтобы получить номер версии последней фиксации, записанной текущим SparkSession, для всех потоков и всех таблиц, выполните запрос конфигурации SQL spark.databricks.delta.lastCommitVersionInSession.

SQL

SET spark.databricks.delta.lastCommitVersionInSession

Питон

spark.conf.get("spark.databricks.delta.lastCommitVersionInSession")

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

spark.conf.get("spark.databricks.delta.lastCommitVersionInSession")

Если SparkSession не сделал коммиты, запрос ключа возвращает пустое значение.