Azure Databricks 上的隔离级别和写入冲突

表的隔离级别定义必须从并发操作所作修改中隔离事务的程度。 Azure Databricks 上的写入冲突取决于隔离级别。

Delta Lake 提供读写之间的 ACID 事务保证。 这表示：

• 多个群集中的多个写入者可能同时修改某个表分区。 写入者会看到该表的一致快照视图，写入将按串行顺序发生。
  • 读者可以继续查看 Azure Databricks 作业开始时使用的表的一致性快照视图，即使在作业过程中修改了某个表也是如此。

请参阅 Azure Databricks 上的 ACID 保证是什么？。

-- Set a table property.
ALTER TABLE table-name SET TBLPROPERTIES ('delta.isolationLevel' = 'Serializable')

-- Enable a feature using a table property and update the table protocol.
ALTER TABLE table_name SET TBLPROPERTIES ('delta.enableDeletionVectors' = true);

-- Drop a table feature.
ALTER TABLE table_name DROP FEATURE deletionVectors;

-- Upgrade to UniForm.
REORG TABLE table_name APPLY (UPGRADE UNIFORM(ICEBERG_COMPAT_VERSION=2));

-- Update the table schema.
ALTER TABLE table_name ADD COLUMNS (col_name STRING);

写入与行级别并发冲突

行级并发可检测行级更改，并自动解决当并发写入操作更新或删除同一数据文件中的不同行时发生的冲突，因而可以降低并发写入操作之间的冲突。

行级并发在 Databricks Runtime 14.2 及更高版本上已正式发布。 对于以下条件，默认情况下支持行级并发：

• 启用了删除向量且未分区的表。
• 使用液态聚类的表，除非已禁用删除向量。

具有分区的表不支持行级并发，但当启用了删除矢量时，仍可以避免 OPTIMIZE 与所有其他写入操作之间的冲突。 请参阅行级别并发的限制。

对于其他 Databricks Runtime 版本，请参阅行级并发预览行为（旧版）。

对行级并发的 MERGE INTO 支持需要 Databricks Runtime 14.2 中的 Photon。 在 Databricks Runtime 14.3 LTS 及更高版本中，不需要 Photon。

下表说明了在每个启用了行级别并发的隔离级别可能出现冲突的操作对。

在没有行级别并发的情况下发生写入冲突

下表说明了在每个隔离级别可能出现冲突的操作对。

如果表定义了分区或未启用删除矢量，则表不支持行级并发。 要实现行级并发，需要 Databricks Runtime 14.2 或更高版本。

行级别并发的限制

行级别并发存在一些限制。 进行以下操作时，冲突解决方法遵循 Azure Databricks 上写入冲突的普通并发。 请参阅在没有行级别并发的情况下发生写入冲突。

• 具有复杂条件子句的命令，包括以下命令：
  • 针对结构、数组或映射等复杂数据类型的命令。
  • 使用非确定性表达式和子查询的条件。
  • 包含关联子查询的条件。
• 在 Databricks Runtime 14.2 中，对于 MERGE 命令，必须对目标表使用显式谓词来筛选与源表匹配的行。 对于合并解决方法，筛选器仅会扫描根据并发操作中的筛选器条件可能会冲突的行。

删除向量的所有限制也适用。 请参阅限制。

Delta Lake 何时在不读取表的情况下提交？

如果满足以下条件，Delta Lake INSERT 或追加操作在提交之前不会读取表状态：

1. 逻辑是使用 INSERT SQL 逻辑或追加模式表示的。
2. 逻辑不包含引用写入操作所针对的表的子查询或条件。

与其他提交一样，Delta Lake 在提交时将使用事务日志中的元数据来验证和解析表版本，但不会实际读取表的版本。

写可序列化与可序列化隔离级别

表的隔离级别定义了必须将某事务与并发事务所作修改进行隔离的程度。 Azure Databricks 上的 Delta Lake 支持两个隔离级别：Serializable 和 WriteSerializable。

• Serializable：最强隔离级别。 它可确保提交的写入操作和所有读取均可序列化。 只要有一个串行序列一次执行一项操作，且生成与表中所示相同的结果，则可执行这些操作。 对于写入操作，串行序列与表的历史记录中所示完全相同。

• WriteSerializable（默认） ：强度比 Serializable 低的隔离级别。 它仅确保写入操作（而非读取）可序列化。 但是，这仍比快照隔离更安全。 WriteSerializable 是默认的隔离级别，因为对大多数常见操作而言，它使数据一致性和可用性之间达到良好的平衡。

在此模式下，Delta 表的内容可能与表历史记录中所示的操作序列不同。 这是因为此模式允许某些并发写入对（例如操作 X 和 Y）继续执行，这样的话，即使历史记录显示在 X 之后提交了 Y，结果也像在 X 之前执行 Y 一样（即它们之间是可序列化的）。若要禁止这种重新排序，请将表隔离级别设置为 Serializable，以使这些事务失败。

读取操作始终使用快照隔离。 写入隔离级别确定读取者是否有可能看到某个“从未存在”（根据历史记录）的表的快照。

对于 Serializable 级别，读取者始终只会看到符合历史记录的表。 对于 WriteSerializable 级别，读取者可能会看到在增量日志中不存在的表。

例如，请考虑 txn1（长期删除）和 txn2（它插入 txn1 删除的数据）。 txn2 和 txn1 完成，并且它们会按照该顺序记录在历史记录中。 根据历史记录，在 txn2 中插入的数据在表中不应该存在。 对于 Serializable 级别，读取者将永远看不到由 txn2 插入的数据。 但是，对于 WriteSerializable 级别，读取者可能会在某个时间点看到由 txn2 插入的数据。

有关在每个隔离级别中哪些类型的操作可能相互冲突以及可能出现的错误的详细信息，请参阅使用分区和非连续命令条件来避免冲突。

设置隔离级别

使用 ALTER TABLE 命令设置隔离级别。

ALTER TABLE <table-name> SET TBLPROPERTIES ('delta.isolationLevel' = <level-name>)

其中，<level-name> 是 Serializable 或 WriteSerializable。

例如，若要将隔离级别从默认的 WriteSerializable 更改为 Serializable，请运行：

ALTER TABLE <table-name> SET TBLPROPERTIES ('delta.isolationLevel' = 'Serializable')

使用分区和非连续命令条件来避免冲突

在所有标记为“可能出现冲突”的情况下，这两个操作是否会冲突取决于它们是否对同一组文件进行操作。 通过将表分区为与操作条件中使用的列相同的列，可以使两组文件不相交。 例如，如果未按日期对表进行分区，则两个命令 UPDATE table WHERE date > '2010-01-01' ... 和 DELETE table WHERE date < '2010-01-01' 将冲突，因为两者都可以尝试修改同一组文件。 按 date 对表进行分区就可以避免此冲突。 因此，根据命令上常用的条件对表进行分区可以显著减少冲突。 但是，由于存在大量子目录，因此按包含高基数的列对表进行分区可能会导致其他性能问题。

冲突异常

发生事务冲突时，你将观察到以下异常之一：

ConcurrentAppendException

当并发操作在操作读取的同一分区（或未分区表中的任何位置）中添加文件时，会发生此异常。 文件添加操作可能是由 INSERT、DELETE、UPDATE 或 MERGE 操作引起的。

在默认隔离级别为 WriteSerializable 的情况下，通过盲目的 INSERT 操作（即，盲目追加数据而不读取任何数据的操作）添加的文件不会与任何操作冲突，即使它们接触相同的分区（或未分区表中的任何位置）也是如此。 如果隔离级别设置为 Serializable，则盲目追加可能会产生冲突。

通常执行 DELETE、UPDATE 或 MERGE 并发操作时会引发此异常。 尽管并发操作可能会物理上更新不同的分区目录，但其中一个可能会读取另一个分区同时更新的同一分区，从而导致冲突。 可以通过在操作条件中设置显式隔离来避免这种情况。 请看下面的示例。

// Target 'deltaTable' is partitioned by date and country
deltaTable.as("t").merge(
    source.as("s"),
    "s.user_id = t.user_id AND s.date = t.date AND s.country = t.country")
  .whenMatched().updateAll()
  .whenNotMatched().insertAll()
  .execute()

假设你在不同的日期或国家/地区同时运行上述代码。 由于每个作业都在目标 Delta 表上的独立分区上运行，因此你不会遇到任何冲突。 但是，该条件不够明确，可能会扫描整个表，并且可能与更新任何其他分区的并发操作冲突。 相反，你可以重写语句以将特定日期和国家/地区添加到合并条件中，如以下示例所示。

// Target 'deltaTable' is partitioned by date and country
deltaTable.as("t").merge(
    source.as("s"),
    "s.user_id = t.user_id AND s.date = t.date AND s.country = t.country AND t.date = '" + <date> + "' AND t.country = '" + <country> + "'")
  .whenMatched().updateAll()
  .whenNotMatched().insertAll()
  .execute()

现在可以安全地在不同日期和国家/地区同时运行此操作。

ConcurrentDeleteReadException

如果某个并发操作删除了你的操作读取的文件，则会发生此异常。 常见的原因是，DELETE、UPDATE 或 MERGE 操作导致了重写文件。

ConcurrentDeleteDeleteException

如果某个并发操作删除了你的操作也删除的文件，则会发生此异常。 这可能是由于两个并发压缩操作重写相同的文件引起的。

MetadataChangedException

当并发事务更新 Delta 表的元数据时，将发生此异常。 常见原因是进行 ALTER TABLE 操作或写入 Delta 表以更新表的架构。

ConcurrentTransactionException

如果使用同一检查点位置的流式处理查询同时启动多次，并尝试同时写入 Delta 表。 请勿让两个流式处理查询使用相同的检查点位置并同时运行。

ProtocolChangedException

在下列情况下，可能会发生此异常：

• 当 Delta 表升级到新协议版本时。 为使将来的操作成功，你可能需要升级 Databricks Runtime。
• 当多个编写器同时创建或替换表时。
• 当多个编写器同时写入一个空路径时。

有关更多详细信息，请参阅 Azure Databricks 如何管理 Delta Lake 功能兼容性？。

行级并发预览行为（旧版）

本部分介绍了 Databricks Runtime 14.1 及更低版本中行级并发的预览行为。 行级并发始终需要删除矢量。

在 Databricks Runtime 13.3 LTS 及更高版本中，启用了 Liquid 聚类的表会自动启用行级并发。

在 Databricks Runtime 14.0 和 14.1 中，可以通过为群集或 SparkSession 设置以下配置，为具有删除矢量的表启用行级并发：

spark.databricks.delta.rowLevelConcurrencyPreview = true

在 Databricks Runtime 14.1 及更低版本中，非 Photon 计算仅支持 DELETE 操作的行级并发。