如何在PySpark DataFrame追加新行时保证unique_id唯一？

临时解决方案（基于现有monotonically_increasing_id生成的ID） #

如果不想修改现有df_a的ID体系，可以通过偏移df_b的生成ID来避免冲突：

1. 先获取df_a中unique_id的最大值
2. 给df_b生成的monotonically_increasing_id加上该最大值+1，确保所有新ID都大于现有ID

import pyspark.sql.functions as F

# 获取df_a的最大unique_id，处理空df_a的情况
max_id_row = df_a.agg(F.max("unique_id")).first()
max_id = max_id_row[0] if max_id_row[0] is not None else 0

# 为df_b生成偏移后的唯一ID
df_b = df_b.withColumn("unique_id", F.monotonically_increasing_id() + (max_id + 1))

# 合并两个DataFrame
combined_df = df_a.union(df_b)
combined_df.show()

这个方案的优势是不需要改动现有df_a的ID，适合临时追加场景；但每次追加都需要计算当前最大ID，且如果后续多次追加，需要重复执行该逻辑。

长期可行解决方案（重新分配df_a的ID体系） #

如果要长期稳定地定期追加数据，建议一次性重新给df_a分配更可控的唯一ID，之后每次追加数据时按规则生成新ID，避免依赖monotonically_increasing_id的不确定性。

方案1：数值型自增ID #

适合需要数值ID且希望ID连续/有序的场景：

第一步：一次性重新分配df_a的ID

from pyspark.sql.window import Window

# 用窗口函数生成连续自增ID（如果df_a数据量极大，可改用分区+本地ID的方式优化性能）
global_window = Window.orderBy("unique_id")  # 按原ID排序保留原有顺序
df_a = df_a.withColumn("unique_id", F.row_number().over(global_window).cast("bigint"))

后续追加数据时的逻辑

每次追加前获取当前最大ID，给新数据生成从max_id + 1开始的ID：

# 获取当前最大ID
max_id = df_a.agg(F.max("unique_id")).first()[0]

# 为df_b生成自增ID（用分区内row_number避免全局排序的性能问题）
partition_window = Window.partitionBy(F.spark_partition_id()).orderBy(F.monotonically_increasing_id())
df_b = df_b.withColumn("unique_id", (F.row_number().over(partition_window) + max_id).cast("bigint"))

# 合并并更新df_a
df_a = df_a.union(df_b)

方案2：UUID字符串ID #

适合不需要数值ID、追求绝对唯一且无需维护计数器的场景：

第一步：一次性替换df_a的ID

df_a = df_a.withColumn("unique_id", F.expr("uuid()"))

后续追加数据时的逻辑

直接给df_b生成UUID即可，无需担心冲突：

df_b = df_b.withColumn("unique_id", F.expr("uuid()"))
df_a = df_a.union(df_b)

UUID方案的优势是实现简单，无需计算最大ID或维护状态；缺点是ID是字符串类型，可能占用更多存储，且无法体现顺序。

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