Используйте жидкую кластеризацию для таблиц

Метод "Liquid clustering" — это техника оптимизации макета данных, которая заменяет секционирование таблиц и ZORDER. Это упрощает управление таблицами и оптимизирует производительность запросов путем автоматической организации данных на основе ключей кластеризации.

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

Это важно

Жидкая кластеризация общедоступна для таблиц Delta Lake в Databricks Runtime 15.4 LTS и более поздних версиях, а для таблиц Apache Iceberg — в рамках общедоступной предварительной версии в Databricks Runtime 16.4 LTS и более поздних версиях. Databricks рекомендует использовать последнюю версию Databricks Runtime для оптимальной производительности.

Управляемые таблицы Apache Iceberg версии 3 также поддерживают векторы удаления, отслеживание строк, параллелизм на уровне строк и автоматическую кластеризацию жидкости. Для этих возможностей требуется Databricks Runtime 18.0 и более поздних версий. Ознакомьтесь с функциями Apache Iceberg версии 3.

Когда следует использовать кластеризацию жидкости

Databricks рекомендует кластеризацию жидкости для всех новых таблиц, включая потоковые таблицы и материализованные представления. Следующие сценарии особенно полезны для кластеризации:

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

Включение кластеризации жидкости

Вы можете включить кластеризацию жидкости в существующей непартиментной таблице или во время создания таблицы. Кластеризация несовместима с секционированием или ZORDER. Databricks рекомендует позволить платформе управлять всеми операциями по макету и оптимизации данных в вашей таблице. После включения liquid-кластеризации запустите OPTIMIZE задания для пошагового кластерирования данных. Узнайте , как активировать кластеризацию.

Создание таблиц с помощью кластеризации

Чтобы включить кластеризацию жидкости, добавьте фразу CLUSTER BY в инструкцию создания таблицы, как показано в приведенных ниже примерах. В Databricks Runtime 14.3 LTS и более поздних версиях можно использовать API кадра данных и API DeltaTable в Python или Scala, чтобы включить кластеризацию жидкости для таблиц Delta Lake.

SQL

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

CREATE TABLE table1 (col0 INT, col1 STRING) CLUSTER BY (col0);

Чтобы создать таблицу из существующих данных с кластерированием, CLUSTER BY необходимо появиться после имени таблицы, а не в предложении SELECT :

CREATE TABLE table2 CLUSTER BY (col0)
AS SELECT * FROM table1;

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

CREATE TABLE table3 LIKE table1;

Python

Чтобы создать пустую таблицу с кластерированием с помощью DeltaTable API:

(DeltaTable.create()
  .tableName("table1")
  .addColumn("col0", dataType = "INT")
  .addColumn("col1", dataType = "STRING")
  .clusterBy("col0")
  .execute())

Чтобы создать таблицу из существующего DataFrame:

df = spark.read.table("table1")
df.write.clusterBy("col0").saveAsTable("table2")

Создание таблицы с помощью DataFrameWriterV2 API (доступно в Databricks Runtime 14.2 и более поздних версиях):

df = spark.read.table("table1")
df.writeTo("table1").using("delta").clusterBy("col0").create()

Scala

Чтобы создать пустую таблицу с кластерированием с помощью DeltaTable API:

DeltaTable.create()
  .tableName("table1")
  .addColumn("col0", dataType = "INT")
  .addColumn("col1", dataType = "STRING")
  .clusterBy("col0")
  .execute()

Чтобы создать таблицу из существующего DataFrame:

val df = spark.read.table("table1")
df.write.clusterBy("col0").saveAsTable("table2")

Создание таблицы с помощью DataFrameWriterV2 API (доступно в Databricks Runtime 14.2 и более поздних версиях):

val df = spark.read.table("table1")
df.writeTo("table1").using("delta").clusterBy("col0").create()

Это важно

При использовании API DataFrame для задания ключей кластеризации можно указать только столбцы кластеризации во время создания таблицы или при использовании режима overwrite (например, с операциями CREATE OR REPLACE TABLE). При использовании append режима невозможно изменить ключи кластеризации.

Чтобы изменить ключи кластеризации в существующей таблице при добавлении данных, используйте команды SQL ALTER TABLE для изменения конфигурации кластеризации отдельно от операций записи данных. См. раздел "Изменение ключей кластеризации".

В Databricks Runtime 16.4 LTS и более поздних версиях можно создавать таблицы с включенной функцией liquid clustering с помощью записи Structured Streaming, как показано в следующих примерах:

SQL

CREATE TABLE table1 (
  col0 STRING,
  col1 DATE,
  col2 BIGINT
)
CLUSTER BY (col0, col1);

Python

(spark.readStream.table("source_table")
  .writeStream
  .clusterBy("column_name")
  .option("checkpointLocation", checkpointPath)
  .toTable("target_table")
)

Scala

spark.readStream.table("source_table")
  .writeStream
  .clusterBy("column_name")
  .option("checkpointLocation", checkpointPath)
  .toTable("target_table")

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

Таблицы Delta Lake с включённой функцией liquid clustering используют Delta writer версии 7 и reader версии 3. Клиенты Delta, которые не поддерживают эти протоколы, не могут получить доступ к этим таблицам. Вы не можете перейти на более раннюю версию протокола таблицы. См. сведения о совместимости функций Delta Lake и протоколах.

Чтобы переопределить включение функций по умолчанию, например векторы удаления, см. раздел "Переопределение функций по умолчанию" (необязательно).

Активировать на существующих таблицах

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

ALTER TABLE <table_name>
CLUSTER BY (<clustering_columns>)

Для управляемых таблиц Apache Iceberg рассмотрим следующее:

  • Для таблиц с спецификацией версии 2 необходимо явно отключить векторы удаления и отслеживание строк при включении кластеризации жидкости в существующей таблице.
  • Для таблиц со спецификацией версии 3 отключение этих функций не требуется, так как поддерживаются векторы удаления и отслеживание строк. Ознакомьтесь с функциями Apache Iceberg версии 3.

Note

Поведение по умолчанию не применяет кластеризацию к ранее записанным данным. Для принудительной перегруппировки используйте OPTIMIZE FULL или OPTIMIZE FULL WHERE <predicate>. См. статью "Принудительное перекластеривание".

Преобразование секционированных таблиц в отказоустойчивую кластеризацию

Начиная с Databricks Runtime 18.1, чтобы перевести существующую секционированную таблицу Delta Lake на жидкую кластеризацию, используйте REPLACE PARTITIONED BY WITH CLUSTER BY в инструкции ALTER TABLE. Преобразование сводит к минимуму время простоя чтения и записи и поддерживает как внешние, так и управляемые таблицы. После преобразования таблица поддерживает чтение с помощью Databricks Runtime 13.3 LTS и более поздних версий.

Note

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

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

  • Улучшения производительности для таблиц, страдающих от неэффективного пропуска данных или чрезмерной партиционированности.
  • Автоматическое повышение производительности с использованием CLUSTER BY AUTO для таблиц с часто меняющимися шаблонами запросов.
  • Кластеризация столбцов является гибкой и простой для изменения, в то время как секционирование жестко и трудно изменить.
  • Уменьшается число конфликтов записи, поскольку таблицы с жидкостной кластеризацией поддерживают параллелизм на уровне строк. См. параллелизм на уровне строк.

Syntax

ALTER TABLE <table_name>
REPLACE PARTITIONED BY WITH CLUSTER BY [( <clustering_columns> ) | AUTO]

Конструкция CLUSTER BY поддерживает следующие параметры:

  • ( <clustering_columns> ): задает новые столбцы кластеризации. Databricks рекомендует сохранить новые столбцы кластеризации, аналогичные исходным столбцам секций. Использование сильно различающихся столбцов запускает масштабную операцию повторной кластеризации при первом запуске OPTIMIZE.
  • AUTO: использует текущие столбцы секционирования в качестве начальных столбцов кластеризации и позволяет с течением времени адаптировать прогнозную оптимизацию. Доступно только для управляемых таблиц каталога Unity. См. автоматическая кластеризация жидкости.
  • Нет указанных параметров: использует текущие столбцы секционирования в качестве новых столбцов кластеризации.

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

Examples

Чтобы выполнить кластеризацию по столбцам, отличным от столбцов исходных разделов, например для таблицы, разделённой по столбцу (year, month, day), сделайте следующее:

ALTER TABLE t1 REPLACE PARTITIONED BY WITH CLUSTER BY (day, id);
OPTIMIZE t1;

Note

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

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

ALTER TABLE t2 REPLACE PARTITIONED BY WITH CLUSTER BY AUTO;

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

ALTER TABLE t3 REPLACE PARTITIONED BY WITH CLUSTER BY;

Обработка параллельных операций чтения и записи во время преобразования

После преобразования Databricks Runtime 13.3 LTS и более поздних версий поддерживается для операций чтения и записи. Azure Databricks рекомендует Databricks Runtime версии 15.4 LTS и выше для рабочих нагрузок, которые считывают или записывают данные в таблицу в процессе преобразования.

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

Тип рабочей нагрузки Операции чтения во время преобразования Операции записи во время преобразования
Batch Без простоев. Все версии среды выполнения Databricks могут считывать таблицу во время преобразования. Нет времени простоя в Databricks Runtime 15.4 и более поздних версий.
Для Databricks Runtime 15.3 и ниже Databricks рекомендует приостановить рабочие нагрузки перед преобразованием, а затем перезапустить рабочие нагрузки после завершения преобразования.
Streaming При использовании отслеживания схемы и сопоставления столбцов: перезапустите поток без потери зафиксированных изменений.
Без отслеживания схемы и сопоставления столбцов: поток вызывает исключение. Перезапустите с новым расположением контрольной точки и начальной версией. Фиксации не теряются.		 Перезапустите поток, не теряя коммиты.

Проверка или откат преобразования

Чтобы подтвердить преобразование, выполните команду DESCRIBE EXTENDED , чтобы просмотреть новые столбцы кластеризации. Выполните команду DESCRIBE HISTORY , чтобы просмотреть ряд операций REORG , UPGRADE PROTOCOL операцию и REPLACE PARTITIONED BY WITH CLUSTER BY операцию.

Чтобы откатить преобразование, используйте RESTORE для возврата к предыдущей версии. Кроме того, можно переписать таблицу с помощью REPLACE TABLE ... PARTITIONED BY (...) AS SELECT * FROM ....

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

ALTER TABLE my_table CLUSTER BY NONE;
ALTER TABLE my_table UNSET TBLPROPERTIES ('delta.liquid.hierarchicalClusteringColumns');
RESTORE TABLE my_table TO VERSION AS OF <version_number_before_conversion>;

См. RESTORE.

Преобразование таблицы, секционированной по столбцу метки времени

Чтобы преобразовать таблицу (t1), секционированную по столбцу метки времени (timestamp_col) и использовать столбец метки времени в качестве ключа кластеризации, необходимо задать дополнительные конфигурации:

SET spark.databricks.delta.liquidConversion.statsGeneration.enabled = false;
ALTER TABLE t1 REPLACE PARTITIONED BY WITH CLUSTER BY (timestamp_col, id);
ANALYZE TABLE t1 COMPUTE DELTA STATISTICS;

Если вы пытаетесь преобразовать столбец секции метки времени в столбец кластеризации без этих конфигураций, команда вызывает ошибку:

ALTER TABLE REPLACE PARTITIONED BY WITH CLUSTER BY cannot auto-generate stats on table with column event_ts due to unsupported type: timestamp. Disable stats auto-generation by setting 'spark.databricks.delta.liquidConversion.statsGeneration.enabled' to 'false' and retry the command again. SQLSTATE: 42000

Ограничения преобразования

Следующие ограничения применяются к команде REPLACE PARTITIONED BY WITH CLUSTER BY преобразования:

  • Потоковые таблицы и материализованные представления, созданные в конвейерах Lakeflow, не поддерживаются. Чтобы использовать жидкую кластеризацию, необходимо обновить определение пайплайна, чтобы использовать CLUSTER BY вместо PARTITIONED BY.
  • Таблицы, использующие Delta Sharing с фильтрацией по разделам, не поддерживаются. Сведения о фильтрации разделов в Delta Sharing см. в разделе Указание разделов таблицы для общего доступа.

Удаление ключей кластеризации

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

ALTER TABLE table_name CLUSTER BY NONE;

Выбор ключей кластеризации

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

Tip

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

Рекомендации по выбору ключей

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

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

Ключи кластеризации должны быть столбцами с собранными статистическими данными. По умолчанию таблицы Delta Lake собирают статистику для первых 32 столбцов. См. раздел "Указание столбцов статистики".

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

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

  • Дата
  • Timestamp
  • TimestampNTZ (Databricks Runtime 14.3 LTS и выше)
  • String
  • Целое число, Длинное, Короткое, Байт
  • Float (Число с плавующей запятой одинарной точности), Double (Число с плавующей запятой двойной точности), Decimal (Десятичное число)

Вы можете группировать по StructField с помощью точечной нотации, например CLUSTER BY (struct_col.field). Вложенные поля структуры поддерживаются для любой глубины, например CLUSTER BY (struct_col.nested.field). Тип данных поля должен быть одним из поддерживаемых типов в предыдущем списке.

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

  • Сложные типы, такие как StructType, MapTypeили ArrayType
  • MapType и ArrayType элементы, такие как map_col['key'], array_col[0]или map_col.key.

Миграция с секционирования или порядка Z

Это важно

Databricks рекомендует использовать автоматическое преобразование с REPLACE PARTITIONED BY WITH CLUSTER BY командой. См. Преобразование секционированной таблицы в жидкую кластеризацию.

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

Текущий метод оптимизации данных Рекомендация по кластеризации ключей
Секционирование в стиле Hive Используйте столбцы секций в качестве ключей кластеризации.
Индексирование Z-порядка ZORDER BY Используйте столбцы в качестве ключей кластеризации.
Секционирование в стиле Hive и порядок Z Используйте как столбцы секционирования, так и ZORDER BY столбцы в качестве ключей кластеризации.
Созданные столбцы для уменьшения кардинальности (например, дата для метки времени) Используйте исходный столбец в качестве ключа кластеризации и не создавайте созданный столбец.

автоматическая кластеризация жидкости

В Databricks Runtime 15.4 LTS и более поздних версиях можно включить автоматическую жидкую кластеризацию для управляемых таблиц Delta Lake в Unity Catalog. Для управляемых таблиц Apache Iceberg версии 3 для каталога Unity для автоматической кластеризации жидкости требуется Databricks Runtime 18.0 и более поздних версий. Автоматическое кластеризация жидкости позволяет Azure Databricks интеллектуально выбирать ключи кластеризации для оптимизации производительности запросов с помощью предложения CLUSTER BY AUTO.

Note

Автоматическая жидкая кластеризация также поддерживается для материализованных представлений и потоковых таблиц, включая конвейеры Lakeflow и автономные конвейеры. Укажите CLUSTER BY AUTO в определении конвейера или SQL.

Как работает автоматическая кластеризация жидкости

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

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

  • Анализ рабочей нагрузки запросов: Azure Databricks анализирует историческую рабочую нагрузку запросов таблицы и определяет оптимальные столбцы для кластеризации.
  • Адаптация к изменениям. Если шаблоны запросов или распределения данных изменяются с течением времени, автоматическое кластеризация жидкости выбирает новые ключи для оптимизации производительности.
  • Выбор с учетом затрат: Azure Databricks изменяет ключи кластеризации только в том случае, если прогнозируемая экономия затрат за счёт улучшенного пропуска данных перевешивает затраты на кластеризацию данных.

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

  • Таблица слишком мала, чтобы извлечь выгоду из объединения жидкостей.
  • Таблица уже имеет эффективную схему кластеризации либо из предыдущих вручную заданных ключей, либо из естественного порядка вставки, который соответствует шаблонам запросов.
  • В таблице нет частых запросов.
  • Вы не используете Databricks Runtime 15.4 LTS или более поздней версии.

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

Совместимость версий среды выполнения Databricks

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

Используйте Databricks Runtime 15.4 LTS или более поздней версии, чтобы гарантировать, что автоматически выбранные ключи используют все ваши рабочие нагрузки и что эти рабочие нагрузки учитываются при выборе новых ключей.

Включение или отключение автоматической кластеризации жидкости

SQL

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

CREATE OR REPLACE TABLE table1 (column01 int, column02 string) CLUSTER BY AUTO;

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

ALTER TABLE table1 CLUSTER BY AUTO;

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

ALTER TABLE table1 CLUSTER BY (c1, c2);
ALTER TABLE table1 CLUSTER BY AUTO;

Кроме того, используйте API Python для задания подсказок в одной операции.

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

ALTER TABLE table1 CLUSTER BY NONE;

Чтобы отключить автоматическое кластеризация жидкости и указать столбцы кластеризации:

ALTER TABLE table1 CLUSTER BY (column01, column02);

Если для существующей таблицы включена автоматическая кластеризация Liquid, выполнение CREATE OR REPLACE table_name без CLUSTER BY AUTO отключает автоматическую кластеризацию и не сохраняет столбцы кластеризации. Чтобы сохранить автоматическую кластеризацию жидкости и все ранее выбранные столбцы, включите CLUSTER BY AUTO в инструкцию replace. С помощью CLUSTER BY AUTOпрогнозной оптимизации используется историческая нагрузка запросов к таблице для определения лучших ключей кластеризации.

Python

API Python доступен в Databricks Runtime 16.4 и выше. Вы можете использовать Python только при создании или замене таблицы. Используйте SQL для изменения clusterByAuto состояния существующей таблицы.

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

df = spark.read.table("table1")
df.write
  .format("delta")
  .option("clusterByAuto", "true")
  .saveAsTable(...)

Чтобы задать начальные подсказки для столбца кластеризации при выборе ключа с помощью DataFrameWriter:

df.write
  .format("delta")
  .clusterBy("clusteringColumn1", "clusteringColumn2")
  .option("clusterByAuto", "true")
  .saveAsTable(...)

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

df.writeTo(...).using("delta")
  .option("clusterByAuto", "true")
  .create()

Чтобы задать начальные подсказки для столбца кластеризации при выборе ключа с помощью DataFrameWriterV2:

df.writeTo(...).using("delta")
  .clusterBy("clusteringColumn1", "clusteringColumn2")
  .option("clusterByAuto", "true")
  .create()

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

spark.readStream.table("source_table")
  .writeStream
  .option("clusterByAuto", "true")
  .option("checkpointLocation", checkpointPath)
  .toTable("target_table")

Чтобы задать начальные подсказки для столбца кластеризации при выборе ключа в потоковой таблице:

spark.readStream.table("source_table")
  .writeStream
  .clusterBy("column1", "column2")
  .option("clusterByAuto", "true")
  .option("checkpointLocation", checkpointPath)
  .toTable("target_table")

При использовании .clusterBy для подсказок при выборе ключа кластера вместе с .option('clusterByAuto', 'true') поведение будет следующим:

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

Это важно

При использовании API clusterByAuto кадра данных параметр можно задать только при использовании overwrite режима. Нельзя задать clusterByAuto при использовании append режима. Это ограничение такое же, как и при ручной настройке столбцов кластеризации. Параметры кластеризации можно настроить только во время создания таблицы или замены с помощью overwrite режима.

В качестве обходного решения, если вы хотите изменить clusterByAuto состояние существующей таблицы при добавлении данных, используйте команды SQL ALTER TABLE для изменения конфигурации кластеризации отдельно от операций записи данных.

Проверьте, включена ли автоматическая кластеризация

Чтобы проверить, включена ли в таблице автоматическая кластеризация жидкости, используйте DESCRIBE TABLE или SHOW TBLPROPERTIES.

Если включена автоматическая кластеризация жидкости, свойству clusterByAuto задано значение true. В свойстве clusteringColumns показаны текущие столбцы кластеризации, которые были автоматически или вручную выбраны.

Ограничения

Автоматическое кластеризация жидкости недоступна для управляемых таблиц Apache Iceberg версии 2. Она поддерживается для управляемых таблиц Apache Iceberg версии 3 в Databricks Runtime 18.0 и более поздних версий.

Запись данных в кластеризованную таблицу

Чтобы записывать данные в таблицу Delta Lake с кластеризацией, необходимо использовать клиент записи Delta, который поддерживает все функции протокола записи Delta для таблиц, используемые при жидкой кластеризации. Для записи в кластеризованную таблицу Iceberg можно использовать Iceberg REST API каталога Unity. В Azure Databricks необходимо использовать Databricks Runtime 13.3 LTS и выше.

Операции, поддерживающие кластеризацию при записи

Операции, которые концентрируются при записи, включают следующие:

  • INSERT INTO операции
  • Выражения CTAS и RTAS
  • COPY INTO из формата Parquet
  • spark.write.mode("append")

Пороговые значения размера для кластеризации

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

Количество столбцов кластеризации Пороговое значение для управляемых таблиц каталога Unity Пороговое значение для других таблиц Delta Lake
1 64 МБ 256 МБ
2 256 МБ 1 ГБ
3 512 МБ 2 ГБ
4 1 ГБ 4 ГБ

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

Рабочие нагрузки в режиме стриминга

Структурированные рабочие нагрузки на потоковые данные поддерживают кластеризацию данных при записи при настройке spark.databricks.delta.liquid.eagerClustering.streaming.enabledtrue конфигурации Spark. Кластеризация для этих рабочих нагрузок активируется только в том случае, если по крайней мере одно из последних пяти обновлений потоковой передачи превышает пороговое значение размера из приведенной выше таблицы.

Как запустить кластеризацию

Прогнозная оптимизация автоматически выполняет OPTIMIZE команды для включенных таблиц. См. прогнозную оптимизацию для управляемых каталогом Unity таблиц. При использовании прогнозной оптимизации Databricks рекомендует отключить все запланированные OPTIMIZE задания.

Чтобы активировать кластеризацию, необходимо использовать Databricks Runtime 13.3 LTS или более поздней версии. Databricks рекомендует Databricks Runtime 17.3 LTS и выше для повышения OPTIMIZE производительности больших таблиц. Используйте команду OPTIMIZE на вашей таблице.

OPTIMIZE table_name;

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

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

Принудительное перекластерирование

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

OPTIMIZE table_name FULL;

Это важно

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

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

Частичная перекластеризация

В Databricks Runtime 18.1 и более поздних версиях можно принудительно выполнить перекластеризацию для подмножества записей с помощью OPTIMIZE FULL WHERE <predicate>. Файл включается, если любая часть его диапазона перекрывается с предикатом. См. параметры.

OPTIMIZE events FULL WHERE event_date >= '2025-01-01';

Чтение данных из кластеризованной таблицы

Данные можно считывать в кластеризованной таблице Delta Lake с помощью любого клиента Delta Lake, поддерживающего чтение векторов удаления. С помощью API REST каталога Iceberg можно считывать данные в кластеризованной таблице Iceberg. Жидкостная кластеризация повышает производительность запросов с помощью автоматического пропуска данных, когда фильтрация осуществляется по ключам кластеризации.

SELECT * FROM table_name WHERE cluster_key_column_name = "some_value";

Управление ключами кластеризации

Узнайте, как кластеризована таблица

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

DESCRIBE TABLE table_name;

DESCRIBE DETAIL table_name;

Изменение ключей кластеризации

Ключи кластеризации для таблицы можно изменить в любое время, выполнив ALTER TABLE команду, как показано в следующем примере:

ALTER TABLE table_name CLUSTER BY (new_column1, new_column2);

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

Кроме того, можно отключить кластеризацию, задав ключи NONEследующим образом:

ALTER TABLE table_name CLUSTER BY NONE;

Установка ключей кластера на NONE не перезаписывает кластеризованные данные, но не позволяет будущим операциям OPTIMIZE использовать ключи кластера.

Использование мягкой кластеризации из внешнего движка

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

CREATE OR REPLACE TABLE main.schema.icebergTable
PARTITIONED BY c1;

Чтобы отключить кластеризацию жидкости, выполните приведенные действия.

ALTER TABLE main.schema.icebergTable DROP PARTITION FIELD c2;

Чтобы изменить ключи кластеризации с помощью эволюции разделов Iceberg:

ALTER TABLE main.schema.icebergTable ADD PARTITION FIELD c2;

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

CREATE OR REPLACE TABLE main.schema.icebergTable
PARTITIONED BY (bucket(c1, 10));

Совместимость таблиц с гибкой кластеризацией

В кластеризации Liquid используются функции таблицы Delta Lake, требующие определенных версий среды выполнения Databricks для чтения и записи. Таблицы, созданные с использованием liquid clustering в Databricks Runtime 14.3 LTS и более поздних версий, по умолчанию используют контрольные точки V2. С помощью контрольной точки V2 в Databricks Runtime версии 13.3 LTS и выше можно читать и писать таблицы. См. Checkpoint V2.

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

Переопределить включение функции по умолчанию (необязательно)

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

  1. Используйте ALTER TABLE для задания свойства таблицы, которое отключает одну или несколько функций. Например, чтобы отключить векторы удаления, выполните следующие действия:

    ALTER TABLE table_name SET TBLPROPERTIES ('delta.enableDeletionVectors' = false);

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

    ALTER TABLE <table_name>
CLUSTER BY (<clustering_columns>)

В следующей таблице содержатся сведения о функциях Delta, которые можно переопределить и как включение влияет на совместимость с версиями среды выполнения Databricks:

Функция Delta Совместимость среды выполнения Свойство для переопределения активации Влияние на кластеризацию жидкости при отключении
Векторы удаления Для чтения и записи требуется Databricks Runtime 12.2 LTS и более поздних версий. 'delta.enableDeletionVectors' = false Отключение векторов удаления также отключает параллелизм на уровне строк, что делает транзакции и операции кластеризации более вероятными для конфликта. См. параллелизм на уровне строк.
DELETE, MERGEи UPDATE команды могут выполняться медленнее.
Отслеживание строк данных Записи требуют Databricks Runtime 13.3 LTS и более поздних версий. Можно считывать из любой версии Databricks Runtime. 'delta.enableRowTracking' = false Отключение отслеживания строк также отключает параллелизм на уровне строк, что делает транзакции и операции кластеризации более вероятными для конфликта. См. параллелизм на уровне строк.
Контрольная точка версии 2 Для чтения и записи требуется Databricks Runtime 13.3 LTS и более поздних версий. 'delta.checkpointPolicy' = 'classic' Не влияет на поведение кластеризации жидкости. См. Checkpoint V2.

Ограничения

  • Databricks Runtime 15.1 и ниже: кластеризация при записи не поддерживает исходные запросы, включающие фильтры, соединения или агрегаты.
  • Databricks Runtime 15.4 LTS и ниже: Вы не можете создать таблицу с включенной поддержкой liquid clustering при записи данных с использованием Structured Streaming. Структурированная потоковая передача можно использовать для записи данных в существующую таблицу с включенным кластеризированием жидкости.
  • Apache Iceberg версии 2. Параллелизм на уровне строк не поддерживается в управляемых таблицах Apache Iceberg версии 2, так как векторы удаления и отслеживание строк не поддерживаются.
    • Параллелизм на уровне строк поддерживается в управляемых таблицах Apache Iceberg версии 3, так как спецификация версии 3 поддерживает векторы удаления и отслеживание строк. Ознакомьтесь с функциями Apache Iceberg версии 3.
  • Last updated on