如何在Databricks中创建版本自增的水印表及调用版本值

1. 创建水印表 #

推荐使用Delta Lake格式（Databricks原生支持，具备ACID事务特性，适合这类需要原子更新的场景），创建仅含version列的表并初始化值为1：

方式一：SQL语句 #

-- 创建Delta格式的水印表
CREATE TABLE IF NOT EXISTS watermark_version (
  version INT
) USING DELTA;

-- 仅在表为空时插入初始值1
INSERT INTO watermark_version
SELECT 1 WHERE NOT EXISTS (SELECT * FROM watermark_version);

方式二：PySpark代码 #

from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# 构造初始数据
initial_df = spark.createDataFrame([(1,)], ["version"])

# 判断表是否存在，不存在则创建；存在且为空则插入初始值
if spark.catalog.tableExists("watermark_version"):
    row_count = spark.sql("SELECT COUNT(*) FROM watermark_version").first()[0]
    if row_count == 0:
        initial_df.write.mode("append").saveAsTable("watermark_version")
else:
    initial_df.write.mode("overwrite").saveAsTable("watermark_version")

2. 脚本运行前获取当前版本值 #

在业务代码开始前，读取表中的版本号供逻辑使用：

# 获取当前版本值
current_version = spark.sql("SELECT version FROM watermark_version").first()[0]
print(f"当前运行版本: {current_version}")

# 后续业务逻辑可直接使用current_version变量

3. 脚本运行完成后更新版本值 #

用原子更新操作将版本号加1，避免并发场景下的版本混乱：

方式一：SQL语句 #

-- 原子性更新版本号（加1）
UPDATE watermark_version SET version = version + 1;

方式二：PySpark代码 #

# 执行版本更新
spark.sql("UPDATE watermark_version SET version = version + 1")
# 刷新表元数据确保读取最新值
spark.catalog.refreshTable("watermark_version")

复用封装（可选） #

如果多个脚本需要使用该逻辑，可以封装成函数：

def get_current_version():
    return spark.sql("SELECT version FROM watermark_version").first()[0]

def increment_version():
    spark.sql("UPDATE watermark_version SET version = version + 1")
    spark.catalog.refreshTable("watermark_version")

# 使用示例
current_v = get_current_version()
# 业务逻辑执行...
increment_version()

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