如何在Azure Databricks的Delta表中生成连续IDENTITY列

我尝试创建带有连续IDENTITY列的Delta表，目的是让客户检查是否存在未接收的数据。但生成的IDENTITY列并非连续，且未按查询中指定的从1开始，这让“INCREMENT BY 1”显得有误导性。根据文档说明，IDENTITY值仅保证唯一但不保证连续，想找可行的替代方案，曾考虑过MERGE后使用ROW_NUMBER，但担心开销过大。

现有代码 #

Python 数据准备 #

store_visitor_type_name = ["apple","peach","banana","mango","ananas"]
card_type_name = ["door","desk","light","coach","sink"]
store_visitor_type_desc = ["monday","tuesday","wednesday","thursday","friday"]
colnames = ["column2","column3","column4"]
data_frame = spark.createDataFrame(zip(store_visitor_type_name,card_type_name,store_visitor_type_desc),colnames)
data_frame.createOrReplaceTempView('vw_increment')
data_frame.display()

SQL Delta表创建与MERGE操作 #

CREATE or REPLACE TABLE TEST(
  `column1SK` BIGINT GENERATED ALWAYS AS IDENTITY (START WITH 1 INCREMENT BY 1)
  ,`column2` STRING
  ,`column3` STRING
  ,`column4` STRING
  ,`inserted_timestamp` TIMESTAMP
  ,`modified_timestamp` TIMESTAMP
)
USING delta
LOCATION '/mnt/Marketing/Sales';


MERGE INTO TEST as target
USING vw_increment as source
ON target.`column2` = source.`column2`

WHEN MATCHED 

AND (target.`column3` <> source.`column3`
OR target.`column4` <> source.`column4`)
THEN
  UPDATE SET
    `column2` = source.`column2`
    ,`modified_timestamp` = current_timestamp() 
    
WHEN NOT MATCHED THEN 
  INSERT (
    `column2`
    ,`column3`
    ,`column4`
    ,`modified_timestamp`
    ,`inserted_timestamp`
  ) VALUES (
    source.`column2`
    ,source.`column3`
    ,source.`column4`
    ,current_timestamp()
    ,current_timestamp()
  )

可行替代方案 #

1. 基于当前最大SK值生成连续ID（小到中等数据量适用） #

每次MERGE前查询表中最大的column1SK值，以此为基础给新增数据分配连续ID，避免全表重新计算ROW_NUMBER：

-- 先获取当前最大SK值，表为空时设为0
SET max_sk = (SELECT IFNULL(MAX(column1SK), 0) FROM TEST);

-- 给源数据新增连续ID列
CREATE OR REPLACE TEMP VIEW vw_increment_with_id AS
SELECT 
  ${max_sk} + ROW_NUMBER() OVER (ORDER BY column2) AS column1SK,
  column2,
  column3,
  column4
FROM vw_increment;

-- 执行MERGE，插入时指定生成的连续ID
MERGE INTO TEST as target
USING vw_increment_with_id as source
ON target.`column2` = source.`column2`

WHEN MATCHED 
AND (target.`column3` <> source.`column3` OR target.`column4` <> source.`column4`)
THEN
  UPDATE SET
    `column3` = source.`column3`,
    `column4` = source.`column4`,
    `modified_timestamp` = current_timestamp() 
    
WHEN NOT MATCHED THEN 
  INSERT (
    `column1SK`,
    `column2`,
    `column3`,
    `column4`,
    `modified_timestamp`,
    `inserted_timestamp`
  ) VALUES (
    source.`column1SK`,
    source.`column2`,
    source.`column3`,
    source.`column4`,
    current_timestamp(),
    current_timestamp()
  );

• 优点：实现简单，仅针对新增数据计算ID，开销远低于全表ROW_NUMBER
• 注意：并发写入场景下需加锁或用事务保证max_sk取值的唯一性，避免ID重复

2. 批次化ID生成（大数据量适用） #

如果数据量极大，全表查询最大SK也有开销，可以按同步批次划分：

• 每次同步时生成唯一批次号，给该批次内的数据分配从批次起始ID开始的连续ID
• 批次起始ID可通过维护一个单独的元数据表来记录，每次同步前更新元数据表获取下一批次的起始值
• 这种方式避免了全表扫描，仅需操作元数据表，并发场景更友好

3. 换用业务层面的数据校验方案（优先推荐） #

既然核心需求是让客户检查未接收数据，不一定非要依赖连续ID：

• 批次号校验：每次同步记录唯一批次号，客户通过核对批次号是否连续来判断是否有遗漏
• 数据哈希校验：对源数据的关键列生成哈希值，同步时将哈希值存入目标表，客户通过对比源和目标的哈希值集合来校验数据完整性
• 计数校验：每次同步记录源数据的行数，客户对比目标表新增行数与源数据行数是否一致

这些方案避开了连续ID的实现难点，开销更低，更贴合实际业务需求。

内容的提问来源于stack exchange，提问作者OrganicMustard