Spark Scala实现DataFrame新增行基于最大序号+1生成自增ID

这个需求在Spark Scala里处理得分场景来看，我整理了几种常用的实现方式，你可以根据自己的并发情况和数据规模来选：

如果是单条插入，或者并发量很低不会有多个请求同时操作的情况，可以直接先获取现有数据的最大row_id，再加1作为新行的ID：

import org.apache.spark.sql.functions._

// 加载现有表数据
val existingDF = spark.table("your_employee_table")

// 获取当前最大row_id，空表时默认返回0（避免空指针）
val maxRowId = existingDF
  .select(coalesce(max("row_id"), lit(0)))
  .head()
  .getLong(0)

// 构造新行数据
val newRowDF = Seq((maxRowId + 1, 33, 5000))
  .toDF("row_id", "emp_id", "sal")

// 合并现有数据与新行，追加写入原表
val updatedDF = existingDF.union(newRowDF)
updatedDF.write.mode("append").saveAsTable("your_employee_table")

注意：这种方式在高并发场景下会有问题——如果两个作业同时读取到同一个maxRowId，插入后就会出现重复的row_id，所以只适合低并发或单线程操作的场景。

如果要一次性插入多条新数据，需要给它们分配连续的row_id，可以用row_number()窗口函数来生成连续ID：

import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window

// 加载现有表数据
val existingDF = spark.table("your_employee_table")
val maxRowId = existingDF.select(coalesce(max("row_id"), lit(0))).head().getLong(0)

// 待插入的新数据（仅包含emp_id和sal，无row_id）
val newEmpData = Seq((44, 6000), (55, 7000), (66, 8000))
  .toDF("emp_id", "sal")

// 用窗口函数生成连续的row_id：从maxRowId+1开始递增
val windowSpec = Window.orderBy("emp_id") // 按emp_id排序保证ID分配稳定
val newDataWithRowId = newEmpData
  .withColumn("row_id", row_number().over(windowSpec) + maxRowId)

// 合并并写入
val updatedDF = existingDF.union(newDataWithRowId)
updatedDF.write.mode("append").saveAsTable("your_employee_table")

这里的orderBy字段可以根据你的业务需求选择，只要保证每次排序结果一致，就能生成连续且不重复的ID。

如果是高并发环境，上面的方法都无法保证ID的唯一性，这时候可以借助外部存储的特性来处理：

3.1 利用数据库自增主键 #

如果你的数据最终写入到支持自增主键的数据库（比如MySQL、PostgreSQL），可以让数据库来自动管理row_id，Spark只需要插入业务字段：

// 待插入的新数据（无需包含row_id）
val newEmpData = Seq((33, 5000), (44, 6000))
  .toDF("emp_id", "sal")

// 写入MySQL，依赖数据库的自增主键
newEmpData.write
  .format("jdbc")
  .option("url", "jdbc:mysql://your_host:3306/your_db")
  .option("dbtable", "your_employee_table")
  .option("user", "db_user")
  .option("password", "db_password")
  .option("useSSL", "false")
  .mode("append")
  .save()

前提：数据库表的row_id字段要设置为自增类型（比如MySQL的AUTO_INCREMENT），这样数据库会自动为每条新插入的记录分配唯一且递增的ID，完全避免并发冲突。

3.2 基于Delta Lake的ACID特性 #

如果用Delta Lake作为存储层，可以利用它的ACID事务特性来保证操作的原子性，避免重复ID：

import io.delta.tables._

// 加载Delta表
val deltaTable = DeltaTable.forPath(spark, "/path/to/your/delta/table")

// 获取当前最大row_id
val maxRowId = deltaTable.toDF
  .select(coalesce(max("row_id"), lit(0)))
  .head()
  .getLong(0)

// 构造新行
val newRowDF = Seq((maxRowId + 1, 33, 5000))
  .toDF("row_id", "emp_id", "sal")

// 使用Merge操作原子性插入（如果row_id已存在则跳过，否则插入）
deltaTable.as("target")
  .merge(newRowDF.as("source"), "target.row_id = source.row_id")
  .whenNotMatchedInsertAll()
  .execute()

Delta Lake的Merge操作是原子性的，即使多个作业同时执行，也能保证不会插入重复的row_id。

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