仅更新已更改的行pyspark增量表databricks

from pyspark.sql import SparkSession

# 创建SparkSession
spark = SparkSession.builder \
    .appName("Incremental Table Update") \
    .getOrCreate()

# 连接数据库并加载表格
df = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:mysql://database_server:port/database_name") \
    .option("dbtable", "table_name") \
    .option("user", "username") \
    .option("password", "password") \
    .load()

from pyspark.sql.functions import col, lit

# 标记更改的行
df = df.withColumn("updated", lit(1))  # 新增一列，表示是否更改

# 检查行中的更改
df = df.withColumn("updated", col("updated").cast("integer"))  # 将列类型转为整数

# 通过关联查询，检测是否有旧数据需要更新
df = df.alias("new").join(df.alias("old"),
                          col("new.primary_key") == col("old.primary_key"), 
                          "leftouter") \
                   .where(col("new.updated") == 1 or col("old.primary_key").isNull()) \
                   .select(col("new.*"))

# 将更新后的数据写回数据库中
df.write \
  .format("jdbc") \
  .option("url", "jdbc:mysql://database_server:port/database_name") \
  .option("dbtable", "table_name") \
  .option("user", "username") \
  .option("password", "password") \
  .option("truncate", "true") \  # 清空表格
  .mode("append") \
  .save()