华为云用户手册

DLI 中Delta的使用限制 仅Spark 3.3.1（3.0.0）及以上版本支持Delta。 DLI支持的Delta版本是Delta 2.3.0。 Spark 3.3.1（3.0.0）版本的SQL仍有部分不支持Delta表的相关开源语法，具体内容请参考表1。 表1 Spark 3.3.1-3.0.0版本SQL不支持的Delta表相关开源语法 不支持的语法 示例 ALTER TABLE REPLACE COLUMNS替换列 alter table table0 replace columns(id1 int,name1 string); SHOW CREATE TABLE显示建表语句 show create table table1; INSERT INTO/OVERWRITE指定静态分区插入表 insert into table1 partition(part='part1') select * from table2; ALTER TABLE ADD/DROP PARTITION管理分区 alter table test_delta_parts1 add partition('2024-10-28'); CONVERT TO DELTA语法暂不支持parquet.`tablePath`格式的表 convert to delta parquet.`obs://bucket0/db0/table0`;

系统响应 返回表的历史操作记录，结果指标代表含义见下表。 表1 结果指标说明 指标名称 指标含义 version 对表操作的版本号 timestamp 当前版本操作的时间戳 userId 当前版本操作的用户id userName 当前版本操作的用户名 operation 操作名称（WRITE|CREATE TABLE|UPDATE|DELETE|MERGE|RESTORE等） operationParameters 操作参数 job 运行该操作的作业的详细信息 notebook 运行操作的笔记的详细信息 clusterId 集群id readVersion 为执行写操作而读取的表的版本 isolationLevel 隔离级别 isBlindAppend 是否追加数据 operationMetrics 操作的度量（例如，修改的文件数、行数、字节数等信息） engineInfo Spark和Delta版本信息

注意事项 如果您的表已经在所需的协议版本上，需要先执行如下语句才能修改成功： ALTER TABLE table_name SET TBLPROPERTIES ('delta.columnMapping.mode' = 'name')； 如果您的表不在所需的协议版本上，需要先执行如下语句才能修改成功： ALTER TABLE table_name SET TBLPROPERTIES ('delta.columnMapping.mode' = 'name','delta.minReaderVersion' = '2','delta.minWriterVersion' = '5')

示例 创建分区表 create table if not exists delta_table0using deltapartitioned by (dt)location 'obs://bucket_name0/db0/delta_table0'asselect 1 as id, 'a1' as name, 10 as price, 1000 as dt; 创建非分区表 create table if not exists delta_table0using deltalocation 'obs://bucket_name0/db0/delta_table0'asselect 1 as id, 'a1' as name, 10 as price;create table delta.`obs://bucket_name0/db0/delta_table0`using deltapartitioned by (part_col1, part_col2)as select id,name,year,class_name from table1 where part_col1=2024;

命令格式 还原Delta表到历史某一时刻的状态： RESTORE [TABLE] [database_name.]table_name|DELTA.`obs_path` [TO] TIMESTAMP AS OF timestamp_expression 还原Delta表到某一历史版本的状态： RESTORE [TABLE] [database_name.]table_name|DELTA.`obs_path` [TO] VERSION AS OF version_code

参数描述 表1 参数描述 参数 描述 database_name Database名称，由字母、数字和下划线（_）组成。 table_name Database中的表名，由字母、数字和下划线（_）组成。 obs_path Obs路径，表示Delta表的存储位置。 timestamp_expression 时间戳，不能晚于当前时间，格式'yyyy-MM-ddTHH:mm:ss.SSS' version_code 1.3.1中查询结果中的版本号

命令格式 CREATE[ OR REPLACE] TABLE [ IF NOT EXISTS] [database_name.]table_name|DELTA.`obs://bucket_name/tbl_path` USING DELTA [ COMMENT table_comment ] [ PARTITIONED BY (partColumnList) ] [ LOCATION location_path] [ AS query_statement ]

参数描述 表1 CREATE TABLE AS SELECT参数描述 参数 描述 database_name Database名称，由字母、数字和下划线（_）组成。 table_name Database中的表名，由字母、数字和下划线（_）组成。 bucket_name obs桶名称。 tbl_path Delta表在obs桶中的存储位置。 using 参数delta，定义和创建Delta table table_comment 表的描述信息。 location_path Delta表的存储位置，当前版本通过表名创建Delta表时必须指定，且只支持obs路径，指定该路径Delta 表会创建为外表。 query_statement select查询表达式

执行sql报错DLI.0005: There should be at least one partition pruning predicate on partitioned table `777dd`.`test_delta_parts1` 解决方案：在console页面设置中添加参数spark.sql.forcePartitionPredicatesOnPartitionedTable.enabled值为false。

执行rename/drop column报错DLI.0005: DeltaAnalysisException: Column rename is not supported for your Delta table... 解决方案：先执行如下语句，再执行rename。 ALTER TABLE delta_perms1 SET TBLPROPERTIES ( 'delta.columnMapping.mode' = 'name', 'delta.minReaderVersion' = '2', 'delta.minWriterVersion' = '5');

命令格式 通过表名创建Delta表 CREATE[ OR REPLACE] TABLE [ IF NOT EXISTS] [database_name.]table_name [ (columnTypeList)] USING DELTA [ COMMENT table_comment ] [ PARTITIONED BY (partColumnList) ] LOCATION location_path 通过delta.`Obs路径`创建Delta表 CREATE[ OR REPLACE] TABLE [ IF NOT EXISTS] DELTA.`obs://bucket_name/tbl_path` [ (columnTypeList)] USING DELTA [ COMMENT table_comment ] [ PARTITIONED BY (partColumnList) ] 通过表名创建，能通过show tables查到该表，当前版本必须制定location，且只能指定为obs路径；通过delta.`Obs路径`创建，不能通过show tables查到。

执行insert into/overwrite table_name partition(part_key='part_value') select ...报错DLI.0005: DeltaAnalysisException: Partition column `dt` not found in schema [id, name] 根因分析：不支持insert into/overwrite table_name partition(part_key='part_value')语法。

执行vacuum报错DLI.0001: IllegalArgumentException: requirement failed: Are you sure you would like to vacuum files with such a low retention period? 根因分析：RETAIN 保留期过短（少于168小时），需要确认是否能清理该时间前的数据，将无法再回头查看在指定数据保留期之前创建的版本。确认清理可在console页面设置中添加参数spark.databricks.delta.retentionDurationCheck.enabled值为false。

参数描述 表1 CREATE TABLE参数描述 参数 描述 database_name Database名称，由字母、数字和下划线（_）组成。 table_name Database中的表名，由字母、数字和下划线（_）组成。 bucket_name obs桶名称。 tbl_path Delta表在obs桶中的存储位置。 columnTypeList 以逗号分隔的带数据类型的列表。列名由字母、数字和下划线（_）组成。 using 参数delta，定义和创建Delta table。 table_comment 表的描述信息。 partColumnList 分区字段列表，列名取自columnTypeList中的列名。 location_path Delta表的存储位置，当前版本必须指定，且只支持obs路径，指定该路径Delta 表会创建为外表。

参数描述 表1 SHALLOW CLONE参数描述 参数 描述 target_db 目标Database名称，由字母、数字和下划线（_）组成。 target_table 目标表名，由字母、数字和下划线（_）组成。 source_db 源Database名称，由字母、数字和下划线（_）组成。 source_table 源表名，由字母、数字和下划线（_）组成。 bucket_name obs桶名称。 tbl_path Delta表在obs桶中的存储位置。 constraint_name 约束名称。 boolExpression 约束条件表达式。

示例 CREATE OR REPLACE TABLE delta_table1 SHALLOW CLONE delta_table0 LOCATION 'obs://bucket0/db0/delta_table1';CREATE TABLE delta.`obs://bucket0/db0/delta_table1` SHALLOW CLONE delta_table0 VERSION AS OF 10;

参数描述 表1 ADD COLUMNS参数描述 参数 描述 tableName 表名。 col_spec 可由[col_name][col_type][nullable][comment]四部分组成。 col_name：新增列名，必须指定。 暂不支持给嵌套列添加新的子列 col_type：新增列类型，必须指定。 nullable：新增列是否可以为空，可以缺省。 comment：新增列comment，可以缺省。

示例 insert into delta_table0 values(1, 'a1', 20);insert into delta_table0 select 1, 'a1', 20;insert into test_delta_parts1 PARTITION(dt) select id,name,dt from test_delta1;-- insert overwrite tableinsert overwrite table delta_table0 select 1, 'a1', 20;

参数描述 表1 MERGE INTO参数 参数 描述 database_name Database名称，由字母、数字和下划线（_）组成。 table_name Database中的表名，由字母、数字和下划线（_）组成。 bucket_name obs桶名称。 tbl_path Delta表在obs桶中的存储位置。 target_alias 目标表的别名。 sub_query 子查询。 source_alias 源表或源表达式的别名。 merge_condition 将源表或表达式和目标表关联起来的条件 condition 过滤条件，可选。 matched_action 当满足条件时进行Delete或Update操作 not_matched_action 当不满足条件时进行Insert操作

示例 部分字段更新 create table h0(id int, comb int, name string, price int) using delta location 'obs://bucket_name0/db0/h0';create table s0(id int, comb int, name string, price int) using delta location 'obs://bucket_name0/db0/s0';insert into h0 values(1, 1, 'h1', 1);insert into s0 values(1, 1, 's1', 1),(2, 2, 's2', 2);//写法1merge into h0 using s0on h0.id = s0.idwhen matched then update set h0.id = s0.id, h0.comb = s0.comb, price = s0.price * 2;//写法2merge into h0 using s0on h0.id = s0.idwhen matched then update set id = s0.id,name = s0.name,comb = s0.comb + h0.comb,price = s0.price + h0.price; 缺省字段更新和插入 create table h0(id int, comb int, name string, price int, flag boolean) using delta location 'obs://bucket_name0/db0/h0';create table s0(id int, comb int, name string, price int, flag boolean) using delta location 'obs://bucket_name0/db0/s0';insert into h0 values(1, 1, 'h1', 1, false);insert into s0 values(1, 2, 's1', 1, true);insert into s0 values(2, 2, 's2', 2, false);merge into h0 as targetusing (select id, comb, name, price, flag from s0) sourceon target.id = source.idwhen matched then update set *when not matched then insert *; 多条件更新和删除 create table h0(id int, comb int, name string, price int, flag boolean) using delta location 'obs://bucket_name0/db0/h0';create table s0(id int, comb int, name string, price int, flag boolean) using delta location 'obs://bucket_name0/db0/s0';insert into h0 values(1, 1, 'h1', 1, false);insert into h0 values(2, 2, 'h2', 1, false);insert into s0 values(1, 1, 's1', 1, true);insert into s0 values(2, 2, 's2', 2, false);insert into s0 values(3, 3, 's3', 3, false);merge into h0using (select id, comb, name, price, flag from s0) sourceon h0.id = source.idwhen matched and h0.flag = false then update set id = source.id, comb = h0.comb + source.comb, price = source.price * 2when matched and h0.flag = true then deletewhen not matched then insert *;

注意事项 RETAIN num HOURS表示保留期阈值，建议设置为至少 7 天。 如果对 Delta 表运行VACUUM，则将无法再回头查看在指定数据保留期之前创建的版本。 Delta Lake 具有一项安全检查，用于防止运行危险的VACUUM命令，当指定保留期阈值少于168小时时会报错限制该操作。 如果确定指定保留期阈值进行vacuum操作，可通过将 Spark 配置属性 spark.databricks.delta.retentionDurationCheck.enabled 设置为 false 来关闭此安全检查。

命令格式 VACUUM[database_name.]table_name|DELTA.`obs://bucket_name/tbl_path` [RETAIN num HOURS]; 支持通过DRY RUN参数模拟执行vacuum操作，返回vacuum将要删除的文件列表： VACUUM[database_name.]table_name|DELTA.`obs://bucket_name/tbl_path` [RETAIN num HOURS] [DRY RUN];

参数描述 表1 UPDATE参数 参数 描述 database_name Database名称，由字母、数字和下划线（_）组成。 table_name Database中的表名，由字母、数字和下划线（_）组成。 bucket_name obs桶名称。 tbl_path Delta表在obs桶中的存储位置。 column 待更新的目标列。 EXPRESSION 需在目标表中更新的源表列值的表达式。 boolExpression 过滤条件表达式。

参数描述 表1 ADD CONSTRAINT参数描述 参数 描述 database_name Database名称，由字母、数字和下划线（_）组成。 table_name Database中的表名，由字母、数字和下划线（_）组成。 bucket_name obs桶名称。 tbl_path Delta表在obs桶中的存储位置。 constraint_name 约束名称。 boolExpression 约束条件表达式。

系统响应 返回表的详细信息或统计信息。 表2 结果参数描述 参数名 参数含义 format 表的格式，在这里是delta id 表的唯一id name 在metaserver中定义的表名 description 关于表的说明 location 表的存储路径 createdAt 建表时间戳 lastModified 最后一次修改的时间戳 partitionColumns 分区列 numFiles 表的最新版本中的文件个数 sizeInBytes 表的最新快照的大小（以字节为单位） properties 为此表设置的所有属性 minReaderVersion 可以读取该表的最低Reader版本 minWriterVersion 可以写入该表的最低Writer版本

命令格式 还原Delta表到历史某一时刻的状态： RESTORE [TABLE] [database_name.]table_name|DELTA.`obs_path` [TO] TIMESTAMP AS OF timestamp_expression 还原Delta表到某一历史版本的状态： RESTORE [TABLE] [database_name.]table_name|DELTA.`obs_path` [TO] VERSION AS OF version_code

参数描述 表1 还原Delta表版本参数说明 参数 描述 database_name Database名称，由字母、数字和下划线（_）组成。 table_name Database中的表名，由字母、数字和下划线（_）组成。 obs_path Obs路径，表示Delta表的存储位置。 timestamp_expression 时间戳，不能晚于当前时间，格式'yyyy-MM-ddTHH:mm:ss.SSS' version_code 1.3.1中查询结果中的版本号

参数描述 表1 查询Delta表历史版本参数说明 参数 描述 database_name Database名称，由字母、数字和下划线（_）组成。 table_name Database中的表名，由字母、数字和下划线（_）组成。 timestamp_expression 时间戳，不能晚于当前时间，格式'yyyy-MM-ddTHH:mm:ss.SSS' version_code 1.3.1中查询结果中的版本号

规则 Hudi表必须执行Clean。 对于Hudi的MOR、COW表，都需要开启Clean。 Hudi表在写入数据时会自动判断是否需要执行Clean，因为Clean的开关默认打开(hoodie.clean.automatic默认为true)。 Clean操作并不是每次写数据时都会触发，至少需要满足两个条件： Hudi表中需要有旧版本的文件。对于COW表来说，只要保证数据被更新过就一定存在旧版本的文件。对于MOR表来说，要保证数据被更新过并且做过Compaction才能有旧版本的文件。 Hudi表满足hoodie.cleaner.commits.retained设置的阈值。如果是Flink写hudi，则至少提交的checkpoint要超过这个阈值；如果是批写Hudi，则批写次数要超过这个阈值。

所需权限 SQL权限 表2 CONVERT TO DELTA所需权限列表 权限描述 表所在数据库的CREATE_TABLE权限 表的ALTER权限 表的INSERT_INTO_TABLE权限 表的DROP_TABLE权限 细粒度权限：dli:database:createTable, dli:table:alter, dli:table:insertIntoTable, dli:table:dropTable。 由LakeFormation提供的元数据服务，权限配置详见LakeFormation文档。