克隆 Azure Databricks 上的数据表

使用 clone 命令在特定版本中在 Azure Databricks 上创建现有表的副本。 克隆可以是深拷贝，也可以是浅拷贝。

Azure Databricks 还支持克隆 Parquet 和 Apache Iceberg 表。 请参阅将 Parquet 表和 Apache Iceberg 表增量克隆到 Delta Lake。

有关将克隆与 Unity Catalog 结合使用的详细信息，请参阅 Unity Catalog 表的浅表克隆。

克隆类型

以下克隆类型可用：

克隆的元数据包括：架构、分区信息、不变量、可为 null 和 TBLPROPERTIES。 对于深层克隆来说，流和COPY INTO元数据也会被克隆。 未被克隆的元数据包括表说明和用户定义的提交元数据。

Hive 元存储的克隆操作行为

对于已注册到 Hive 元存储中的 Delta Lake 表，或未注册为表的文件集合，克隆操作具有以下行为：

• 对深层或浅表克隆的更改不会影响源表。
• 浅克隆会引用源目录中的数据文件。 如果对源表运行 VACUUM，客户端将无法再读取这些数据文件，这会引发 FileNotFoundException。 若要修复，请在浅克隆上使用 replace 运行 clone 命令。 如果这种情况经常发生，请考虑使用不依赖于源表的深层克隆。
• 深度克隆不依赖于源表，但创建成本高昂，因为它们同时复制数据和元数据。
• 使用 replace 克隆到其在该路径下已存在表的目标时，如果 Delta 日志不存在，则会创建一个 Delta 日志。 运行 VACUUM 以清理任何现有数据。
• 对于现有的 Delta 表，克隆会创建一个新的增量提交，其中仅包含自上次克隆以来源表中的新增元数据和数据。
• 克隆一个表不同于 Create Table As Select（CTAS）。 克隆除了复制数据之外，还会复制源表元数据。 无需指定分区、格式、不变量、可空性或其他设置。
• 克隆的表具有与其源表无关的历史记录。 对克隆表执行时间旅行查询时，不能使用与源表相同的输入参数。

克隆指标

CLONE 在操作完成后以单行数据帧的形式报告以下指标：

• source_table_size：正在克隆的源表的大小（以字节为单位）。
• source_num_of_files：源表中的文件数。
• num_removed_files：从当前表中删除了多少文件（如果正在替换表）。
• num_copied_files：已从源复制的文件数（如果是浅表克隆，则为 0）。
• removed_files_size：从当前表中删除的文件的大小（以字节为单位）。
• copied_files_size：复制到表中的文件的大小（以字节为单位）。

权限

你必须为 Azure Databricks 表访问控制和云提供商配置权限。

表访问控制

深层克隆和浅表克隆都需要以下权限：

• 对源表的 SELECT 权限。
• 如果您使用CLONE创建新表，请确保在要创建该表的数据库中具有CREATE权限。
• 如果使用 CLONE 替换表，则必须对表具有 MODIFY 权限。

云提供商权限

深度克隆的读者需要拥有对该克隆目录的读取权限。 编写器需要对克隆的目录具有写入访问权限。

浅表克隆的读取者需要对源表的数据文件和克隆的目录具有读取访问权限，因为数据文件保留在源中。 编写器需要对克隆的目录具有写入访问权限。

例子

创建深度或浅克隆

以下代码示例演示如何创建深度和浅表克隆：

SQL

创建深层克隆：

CREATE TABLE target_table CLONE source_table;

替换现有目标：

CREATE OR REPLACE TABLE target_table CLONE source_table;

创建深度克隆，或者如果目标已存在则跳过：

CREATE TABLE IF NOT EXISTS target_table CLONE source_table;

基于最新版本、特定版本或特定时间戳创建浅克隆。 时间戳可以是类似 '2019-01-01' 的日期字符串，也可以是类似 date_sub(current_date(), 1) 的表达式。

CREATE TABLE target_table SHALLOW CLONE source_table;

CREATE TABLE target_table SHALLOW CLONE source_table VERSION AS OF version;

CREATE TABLE target_table SHALLOW CLONE source_table TIMESTAMP AS OF timestamp_expression;

Python

Python DeltaTable API 特定于 Delta Lake。

克隆最新版本的源代码：

from delta.tables import *

deltaTable = DeltaTable.forName(spark, "source_table")
deltaTable.clone(target="target_table", isShallow=True, replace=False)

在特定版本克隆源代码：

deltaTable.cloneAtVersion(version=1, target="target_table", isShallow=True, replace=False)

在特定时间戳处克隆源：

deltaTable.cloneAtTimestamp(timestamp="2019-01-01", target="target_table", isShallow=True, replace=False)

Scala(编程语言)

Scala DeltaTable API 特定于 Delta Lake。

按最新版本克隆源代码：

import io.delta.tables._

val deltaTable = DeltaTable.forName(spark, "source_table")
deltaTable.clone(target="target_table", isShallow=true, replace=false)

克隆特定版本的源代码：

deltaTable.cloneAtVersion(version=1, target="target_table", isShallow=true, replace=false)

在特定时间戳处克隆源：

deltaTable.cloneAtTimestamp(timestamp="2019-01-01", target="target_table", isShallow=true, replace=false)

有关语法详细信息，请参阅 CREATE TABLE CLONE。

检查在CLONE期间复制的元数据

此示例确认在 CLONE 操作期间会复制哪些元数据，具体包括 TBLPROPERTIES、Unity Catalog 标记和 Delta Lake 历史记录。

使用自定义属性和非默认日志保留持续时间设置源表，然后插入数据以生成表历史记录：

CREATE OR REPLACE TABLE test_clone_source (id INT, val STRING)
TBLPROPERTIES ('my.custom.prop' = 'hello', 'delta.logRetentionDuration' = '12 days');

ALTER TABLE test_clone_source SET TAGS ('team' = 'data-eng', 'env' = 'prod');
INSERT INTO test_clone_source VALUES (1, 'a');
INSERT INTO test_clone_source VALUES (2, 'b');

创建深度克隆和浅表克隆：

CREATE OR REPLACE TABLE test_clone_deep DEEP CLONE test_clone_source;

CREATE OR REPLACE TABLE test_clone_shallow SHALLOW CLONE test_clone_source;

确认 TBLPROPERTIES 已复制到这两个克隆中：

SHOW TBLPROPERTIES test_clone_source;
SHOW TBLPROPERTIES test_clone_deep;
SHOW TBLPROPERTIES test_clone_shallow;

确认 Unity 目录标记未复制到克隆：

SELECT catalog_name, schema_name, table_name, tag_name, tag_value FROM information_schema.table_tags WHERE table_name = 'test_clone_source';
SELECT catalog_name, schema_name, table_name, tag_name, tag_value FROM information_schema.table_tags WHERE table_name = 'test_clone_deep';
SELECT catalog_name, schema_name, table_name, tag_name, tag_value FROM information_schema.table_tags WHERE table_name = 'test_clone_shallow';

确认 Delta Lake 历史记录不会被复制到克隆中：

DESCRIBE HISTORY test_clone_source;
DESCRIBE HISTORY test_clone_deep;
DESCRIBE HISTORY test_clone_shallow;

收拾：

DROP TABLE IF EXISTS test_clone_shallow;
DROP TABLE IF EXISTS test_clone_source;
DROP TABLE IF EXISTS test_clone_deep;

数据存档

可以使用深度克隆来保留表在某个时间点的状态，用于存档。 可以增量同步深度克隆，以维护源表的更新状态以进行灾难恢复。

每月运行以下命令一次以同步存档：

CREATE OR REPLACE TABLE archive_table CLONE my_prod_table

ML 模型复制

对于机器学习用例，你可能想要存档用于训练 ML 模型的表版本。 可以使用此存档数据集测试将来的模型。 若要使用 CLONE 存档数据集版本，请执行以下操作：

例如，若要将用于训练模型的表的第 15 版归档：

CREATE TABLE model_dataset CLONE entire_dataset VERSION AS OF 15

生产表上的短期试验

若要在不破坏生产数据表的情况下测试工作流，请创建一个浅克隆。 浅克隆允许你在克隆的表上运行工作负载，该表包含所有生产数据，但不会影响任何生产工作负载。

示例

创建生产表的浅表克隆：

CREATE OR REPLACE TABLE my_test SHALLOW CLONE my_prod_table;

在克隆上运行更新和验证：

UPDATE my_test WHERE user_id is null SET invalid=true;

准备就绪后，请将更改合并回去。 合并会利用克隆中的更新信息，在可能的情况下将范围缩减到仅包含已更改的文件：

MERGE INTO my_prod_table
USING my_test
ON my_test.user_id <=> my_prod_table.user_id
WHEN MATCHED AND my_test.user_id is null THEN UPDATE *;

完成后删除克隆：

DROP TABLE my_test;

覆盖表属性

表属性替代特别适用于：

• 在与不同的业务部门共享数据时，使用所有者或用户信息对表进行批注。
• 需要对 Delta 表及其历史记录或“时光回溯”进行存档。 可以单独为存档表指定数据和日志保留期。 例如： 。

SQL

对于 Delta Lake 表：

CREATE OR REPLACE TABLE archive_table CLONE prod.my_table
TBLPROPERTIES (
delta.logRetentionDuration = '3650 days',
delta.deletedFileRetentionDuration = '3650 days'
)

对于 Iceberg 表：

CREATE OR REPLACE TABLE archive_table CLONE prod.my_table
TBLPROPERTIES (
iceberg.logRetentionDuration = '3650 days',
iceberg.deletedFileRetentionDuration = '3650 days'
)

Python

Python DeltaTable API 特定于 Delta Lake。

dt = DeltaTable.forName(spark, "prod.my_table")
tblProps = {
"delta.logRetentionDuration": "3650 days",
"delta.deletedFileRetentionDuration": "3650 days"
}
dt.clone(target="archive_table", isShallow=False, replace=True, tblProps)

Scala(编程语言)

Scala DeltaTable API 特定于 Delta Lake。

val dt = DeltaTable.forName(spark, "prod.my_table")
val tblProps = Map(
"delta.logRetentionDuration" -> "3650 days",
"delta.deletedFileRetentionDuration" -> "3650 days"
)
dt.clone(target="archive_table", isShallow = false, replace = true, properties = tblProps)