使用液态聚类对表进行分析

液体聚类分析是一种数据布局优化技术，用于替换表分区和 ZORDER。 它通过基于群集密钥自动组织数据，简化了表管理和优化查询性能。

与传统分区不同，可以重新定义聚类键，而且无需重写现有数据。 这样，数据布局就可以随着分析需求的变化而发展。 液态聚类适用于流式表和物化视图。

何时使用液体聚类分析

Databricks 建议对所有新表采用液态聚类，包括流式处理表和物化视图。 以下方案特别受益于群集：

• 通常按高基数列筛选的表。
• 数据分布中存在倾斜的表。
• 增长迅速且需要维护和优化工作量的表。
• 具有并发写入要求的表。
• 具有随时间变化的访问模式的表。
• 典型分区键可能使表具有过多或过少的分区的表。

启用 liquid 聚类分析

可以在现有未分区表上或在创建表期间启用液体聚类分析。 聚类分析与分区或 ZORDER 不兼容。 Databricks 建议允许平台管理您表格中数据的所有布局和优化操作。 启用液体聚类分析后，运行 OPTIMIZE 作业以增量方式对数据进行群集化。 请参阅如何触发聚类分析。

使用聚类分析创建表

若要启用液体聚类分析，请将 CLUSTER BY 短语添加到表创建语句，如以下示例所示。 在 Databricks Runtime 14.2 及更高版本中，可以使用 Python 或 Scala 中的数据帧 API 和 DeltaTable API 为 Delta Lake 表启用液体聚类分析。

SQL

-- Create an empty Delta table with clustering on col0
CREATE TABLE table1(col0 INT, col1 string) CLUSTER BY (col0);

-- Create table from existing data with clustering
-- Note: CLUSTER BY must appear after table name, not in SELECT clause
CREATE TABLE table2 CLUSTER BY (col0)
AS SELECT * FROM table1;

-- Copy table structure including clustering configuration
CREATE TABLE table3 LIKE table1;

Python

# Create an empty Delta table with clustering on col0
(DeltaTable.create()
  .tableName("table1")
  .addColumn("col0", dataType = "INT")
  .addColumn("col1", dataType = "STRING")
  .clusterBy("col0")  # Single clustering key
  .execute())

# Create clustered table from existing DataFrame
df = spark.read.table("table1")
df.write.clusterBy("col0").saveAsTable("table2")

# Alternative: DataFrameWriterV2 API (DBR 14.2+)
df = spark.read.table("table1")
df.writeTo("table1").using("delta").clusterBy("col0").create()

Scala(编程语言)

// Create an empty Delta table with clustering on col0
DeltaTable.create()
  .tableName("table1")
  .addColumn("col0", dataType = "INT")
  .addColumn("col1", dataType = "STRING")
  .clusterBy("col0")
  .execute()

// Create clustered table from existing DataFrame
val df = spark.read.table("table1")
df.write.clusterBy("col0").saveAsTable("table2")

// Alternative: DataFrameWriterV2 API (DBR 14.2+)
val df = spark.read.table("table1")
df.writeTo("table1").using("delta").clusterBy("col0").create()

在 Databricks Runtime 16.0 及更高版本中，可以使用结构化流处理写入创建支持动态数据聚类的表。 Databricks 建议使用 Databricks Runtime 16.4 及更高版本实现最佳性能，如以下示例所示：

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 表上启用液体聚类分析：

-- Alter an existing table
ALTER TABLE <table_name>
CLUSTER BY (<clustering_columns>)

对于 Apache Iceberg，在现有托管 Iceberg 表上启用液体聚类分析时，必须显式禁用删除矢量和行 ID。

删除群集密钥

要删除聚类分析键，请使用以下语法：

ALTER TABLE table_name CLUSTER BY NONE;

选择群集键

关键选择指南

手动指定聚类键时，请根据查询筛选器中最常用的列来选择列。 可以按任意顺序定义聚类键。 如果两列高度相关，只需将其中一列做为聚类键。

最多可以指定 四个聚类分析键。 对于较小的表（小于 10 TB），使用更多群集键在筛选单个列时可能会降低性能。 例如，使用四个键进行筛选比使用两个键进行筛选更糟糕。 但是，随着表大小的增加，对于单列查询而言，这种性能差异可以忽略不计。

聚类分析键必须是收集统计信息的列。 默认情况下，Delta 表中的前 32 列收集统计信息。 请参阅指定增量统计信息列。

支持的数据类型

聚类分析支持以下数据类型用作聚类键：

• 日期
• 时间戳
• TimestampNTZ （Databricks Runtime 14.3 LTS 及更高版本）
• 字符串
• 整数、长、短、字节
• Float、Double、Decimal

从分区或 Z 顺序迁移

若要转换现有表，请考虑以下建议：

自动液体聚类分析

在 Databricks Runtime 15.4 LTS 及更高版本中，可以为 Unity 目录托管 Delta 表启用自动液体聚类分析。 使用自动液体聚类功能，Azure Databricks 可以使用 CLUSTER BY AUTO 子句智能地选择聚类键来优化查询性能。

自动液体聚类分析的工作原理

自动液体聚类分析根据使用模式提供智能优化：

• 需要预测优化：自动密钥选择和聚类操作以异步方式作为维护操作运行。

• 分析查询工作负荷：Azure Databricks 分析表的历史查询工作负荷，并标识用于聚类分析的最佳候选列。

• 适应更改：如果查询模式或数据分布随时间而变化，则自动液体聚类分析会选择新的键来优化性能。

• 成本感知选择：Azure Databricks 仅在通过数据跳过改进预测的成本节省超过数据聚类成本时，才会更改聚类键。

由于以下原因，自动液体聚类分析可能不会选择密钥：

• 表太小，无法受益于液体聚类分析。
• 该表已经拥有一个有效的聚类方案，可能来源于之前的手动键，或者是与查询模式匹配的自然插入顺序。
• 该表没有频繁的查询。
• 未使用 Databricks Runtime 15.4 LTS 或更高版本。

无论数据和查询特征如何，都可以为所有 Unity 目录托管表应用自动液体聚类分析。 启发式会决定选择聚类键是否具有成本效益。

DBR 版本兼容性

可以从支持液体聚类分析的所有 Databricks Runtime 版本读取或写入启用了自动聚类分析的表。 但是，智能密钥选择依赖于 Databricks Runtime 15.4 LTS 中引入的元数据。

使用 Databricks Runtime 15.4 LTS 或更高版本来确保自动选择的键有利于所有工作负载，并在选择新键时考虑这些工作负载。

启用或禁用自动液体聚类分析

若要启用或禁用新表或现有表上的自动液体聚类分析，请使用以下语法：

SQL

-- Create an empty table.
CREATE OR REPLACE TABLE table1(column01 int, column02 string) CLUSTER BY AUTO;

-- Enable automatic liquid clustering on an existing table,
-- including tables that previously had manually specified keys.
ALTER TABLE table1 CLUSTER BY AUTO;

-- Disable automatic liquid clustering on an existing table.
ALTER TABLE table1 CLUSTER BY NONE;

-- Disable automatic liquid clustering by setting the clustering keys
-- to chosen clustering columns or new columns.
ALTER TABLE table1 CLUSTER BY (column01, column02);

如果在未指定CREATE OR REPLACE table_name的情况下运行CLUSTER BY AUTO且表已存在且已启用自动液体聚类分析，则AUTO禁用该设置，并且不会保留聚类分析列。 若要保留液体自动聚类以及以前选择的任何聚类列，请在替换语句中包含 CLUSTER BY AUTO。 在保留时，预测性优化会维护表的历史查询工作负荷，以确定最佳聚簇键。

Python

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

# Set clustering columns and auto to provide a hint for initial selection
df.write
  .format("delta")
  .clusterBy("clusteringColumn1", "clusteringColumn2")
  .option("clusterByAuto", "true")
  .saveAsTable(...)

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

# Set clustering columns and auto to provide a hint for initial selection
df.writeTo(...).using("delta")
  .clusterBy("clusteringColumn1", "clusteringColumn2")
  .option("clusterByAuto", "true")
  .create()

# Set clusterByAuto for streaming tables
spark.readStream.table("source_table")
  .writeStream
  .option("clusterByAuto", "true")
  .option("checkpointLocation", checkpointPath)
  .toTable("target_table")

# Specify a hint for clustering columns with both auto and columns
spark.readStream.table("source_table")
  .writeStream
 .clusterBy("column1", "column2")
  .option("clusterByAuto", "true")
  .option("checkpointLocation", checkpointPath)
  .toTable("target_table")

Databricks Runtime 16.4 及更高版本中提供了 Python API。 使用.clusterBy和.option('clusterByAuto', 'true)一起时，行为如下所示：

• 如果这是第一次设置自动液体聚类，则始终尊重手动输入，并设置 .clusterBy 中的聚类列。
• 如果这是一个具有自动液体聚类的表，则可以接受一次使用.clusterBy的提示。 例如，只有在表尚未设置聚集列时，才会设置.clusterBy指定的列。

只能在创建或替换表时使用 Python。 使用 SQL 更改 clusterByAuto 现有表的状态。

检查是否启用了自动聚类分析

若要检查表是否启用了自动液体聚类分析，请使用 DESCRIBE TABLE 或 SHOW TBLPROPERTIES。

如果启用了自动液体聚类分析，则属性 clusterByAuto 设置为 true。 该clusteringColumns 属性显示自动或手动选择的目前的聚类列。

限制

自动液体聚类分析不适用于 Apache Iceberg。

将数据写入聚类分析表

若要写入聚集 Delta 表，必须使用支持液体聚类分析使用的所有 Delta 写入协议表功能的 Delta 编写器客户端。 若要写入集群 Iceberg 表，可以使用 Unity Catalog 的 Iceberg REST Catalog API。 在 Azure Databricks 上，必须使用 Databricks Runtime 13.3 LTS 及更高版本。

支持写入时群集的操作

写入时群集的操作包括：

• INSERT INTO 操作
• CTAS 和 RTAS 语句
• Parquet 格式的 COPY INTO
• spark.write.mode("append")

群集的大小阈值

写入时聚类分析仅在事务中的数据满足大小阈值时触发。 这些阈值因聚类分析列数而异，在 Unity Catalog 托管表中比在其他 Delta 表中更低。

由于并非所有操作都应用 liquid 聚类分析，因此 Databricks 建议经常运行 OPTIMIZE，以确保有效地聚类分析所有数据。

流式处理工作负荷

将 Spark 配置 spark.databricks.delta.liquid.eagerClustering.streaming.enabled 设置为 true 时，结构化流式处理工作负载支持在写入时进行群集处理。 仅当过去五个流式处理更新中的至少一个超出上表中的大小阈值时，才会触发这些工作负荷的聚类分析。

如何触发聚类分析

预测性优化会自动对已启用的表运行 OPTIMIZE 命令。

若要触发聚类分析，必须使用 Databricks Runtime 13.3 LTS 或更高版本。 在您的表上使用 OPTIMIZE 命令：

OPTIMIZE table_name;

液体聚类分析是 增量的，这意味着 OPTIMIZE 仅根据需要重写数据以适应需要聚类分析的数据。 OPTIMIZE 不会使用与所聚集的数据不匹配的群集键重写数据文件。

如果不使用预测优化，Databricks 建议定期调度 OPTIMIZE 作业来对数据进行聚类。 对于经历多次更新或插入的表，Databricks 建议每隔一或两个小时安排一次 OPTIMIZE 作业。 由于 liquid 聚类分析是增量的，因此聚类分析表的大多数 OPTIMIZE 作业运行速度很快。

强制对所有记录重新聚类

在 Databricks Runtime 16.0 及更高版本中，可使用以下语法强制对表中的所有记录重新聚类：

OPTIMIZE table_name FULL;

首次启用群集或更改群集键时运行 OPTIMIZE FULL。 如果之前已运行 OPTIMIZE FULL，并且群集键没有更改，则 OPTIMIZE FULL 的运行方式与 OPTIMIZE 相同。 在此方案中， OPTIMIZE 使用增量方法，仅重写以前未压缩的文件。 始终使用 OPTIMIZE FULL，以确保数据布局反映当前聚类分析键。

从聚类分析表读取数据

可以使用支持读取删除向量的任何 Delta Lake 客户端读取聚集 Delta 表中的数据。 使用 Iceberg REST 目录 API，可以在聚集的 Iceberg 表中读取数据。 当筛选聚类键时，Liquid 聚类通过自动跳过数据来提高查询性能。

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 Catalog 会删除表达式，并将该列用作聚类键：

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

表与 liquid 聚类分析的兼容性

Liquid clustering 使用 Delta 表功能，这些功能需要特定的 Databricks Runtime 版本进行读取和写入。 默认情况下，在 Databricks Runtime 14.1 及更高版本中使用 liquid 聚类分析创建的表会使用 v2 检查点。 可以使用 Databricks Runtime 13.3 LTS 及更高版本中的 v2 检查点对表进行读取和写入。

可以禁用 v2 检查点和降级表协议，以读取 Databricks Runtime 12.2 LTS 及更高版本中具有 liquid 聚类分析的表。 请参阅删除 Delta Lake 表功能并降级表协议。

替代默认功能启用（可选）

可以在启用液体集群功能时，覆盖默认的 Delta 表功能设置。 这可以防止与这些表功能关联的读取器和编写器协议升级。 必须具有现有表才能完成以下步骤：

1. 使用 ALTER TABLE 设置会禁用一个或多个功能的表属性。 例如，若要禁用删除矢量，请运行以下命令：

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

2. 通过运行以下命令，在表上启用 liquid 群集：

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

在下表中了解你可替代的 Delta 功能，并了解启用如何影响与 Databricks Runtime 版本的兼容性。

限制

• DBR 15.1 和更低版本：写时聚类不支持包含筛选器、联接或聚合操作的源查询。
• DBR 15.4 LTS 及更低版本：无法使用结构化流式写入创建启用了液体聚类的表。 可以使用结构化流式处理将数据写入启用了 liquid 聚类分析的现有表。
• Apache Iceberg v2：使用 Apache Iceberg v2 的托管 Iceberg 表不支持行级并发，因为 Iceberg 表不支持删除矢量和行跟踪。