如何在Databricks中生成跨运行的连续序列ID

Spark是无状态计算引擎，每次运行笔记本都会从头计算，所以要实现类似Oracle/SQL Server序列的自增tno，必须持久化保存当前序列的最大值，每次运行时读取这个值作为起始点，再生成新的自增ID。下面是几种可行的实现方式：

方法1：用Delta表存储序列值（生产环境推荐） #

这是最可靠的方案，依托Delta表的ACID特性保证数据一致性：

1. 初始化序列表（仅第一次运行）：

# 创建存储序列最大值的Delta表
sequence_df = spark.createDataFrame([(3,)], ["current_max_tno"])
sequence_df.write.mode("overwrite").format("delta").saveAsTable("sequence_tno")

2. 每次运行笔记本时执行以下逻辑：

# 读取当前序列的最大值
current_max = spark.sql("SELECT current_max_tno FROM sequence_tno").first()[0]

# 假设你的新数据为new_data_df（仅包含data_value列）
from pyspark.sql.window import Window
from pyspark.sql.functions import row_number, lit

# 按需求排序（不影响ID唯一性的话，也可以用其他排序方式）
window_spec = Window.orderBy("data_value")
# 生成自增tno
new_data_with_tno = new_data_df.withColumn(
    "tno",
    lit(current_max) + row_number().over(window_spec)
)

# 更新序列表的最大值
new_max = current_max + new_data_with_tno.count()
spark.sql(f"UPDATE sequence_tno SET current_max_tno = {new_max}")

# 查看结果
new_data_with_tno.show()

多并发场景下，Delta表的ACID特性会自动保证更新的原子性，避免ID重复。

方法2：用DBFS文件存储序列值（简单场景适用） #

不需要创建额外表，适合测试或低并发场景：

1. 初始化存储文件（仅第一次运行）：

# 在DBFS写入初始最大值
dbutils.fs.put("/tmp/sequence_tno.txt", "3", overwrite=True)

2. 每次运行时的逻辑：

# 读取文件中的当前最大值
max_tno_str = dbutils.fs.head("/tmp/sequence_tno.txt")
current_max = int(max_tno_str)

# 生成自增tno
new_data_with_tno = new_data_df.withColumn(
    "tno",
    lit(current_max) + row_number().over(Window.orderBy("data_value"))
)

# 更新文件中的最大值
new_max = current_max + new_data_with_tno.count()
dbutils.fs.put("/tmp/sequence_tno.txt", str(new_max), overwrite=True)

new_data_with_tno.show()

注意：这种方式没有原子性保障，高并发下可能出现ID重复，不建议生产环境使用。

方法3：用全局临时视图存储（仅临时测试） #

全局临时视图的生命周期和集群绑定，集群重启后数据丢失，仅适合临时测试：

# 检查全局视图是否存在，初始化或读取当前最大值
if spark.catalog._jcatalog.tableExists("global_temp.sequence_tno"):
    current_max = spark.sql("SELECT current_max_tno FROM global_temp.sequence_tno").first()[0]
else:
    current_max = 0
    spark.createDataFrame([(current_max,)], ["current_max_tno"]).createGlobalTempView("sequence_tno")

# 生成自增tno
new_data_with_tno = new_data_df.withColumn(
    "tno",
    lit(current_max) + row_number().over(Window.orderBy("data_value"))
)

# 更新视图中的最大值
new_max = current_max + new_data_with_tno.count()
spark.sql(f"CREATE OR REPLACE GLOBAL TEMP VIEW sequence_tno AS SELECT {new_max} AS current_max_tno")

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