比较两个Dataframe并在Pyspark中运行"Update Else Insert“

from pyspark import SparkContext, SparkConf
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, lit

spark = SparkSession.builder \
    .appName("Update Else Insert") \
    .getOrCreate()

# 读取源Dataframe（df1）和目标Dataframe（df2）
df1 = spark.read.option("header", "true").csv("path/to/source.csv")
df2 = spark.read.option("header", "true").csv("path/to/target.csv")

# 指定关联列
join_columns = ["key_column"]

# 使用左外连接（left_outer）将源Dataframe和目标Dataframe关联
# 如果目标Dataframe中不存在匹配的行，则使用NULL填充
joined_df = df1.join(df2, on=join_columns, how="left_outer")

# 筛选出需要更新或插入的数据行
update_rows = joined_df.filter(col("target_column").isNotNull())
insert_rows = joined_df.filter(col("target_column").isNull())

# 使用when-otherwise条件判断进行行级别更新
updated_df = df2.alias("target").join(update_rows.alias("source"), on=join_columns, how="left_outer") \
    .select(
        col("target.key_column"),
        col("source.update_column").alias("target_column")
        # 其他需要更新的列
    ) \
    .withColumn("updated_column", lit("update_value"))  # 更新列的值

# 更新目标Dataframe
df2 = df2.alias("target").join(updated_df, on="key_column", how="left_outer") \
    .select(
        col("target.key_column"),
        col("updated_column").alias("target_column")
        # 其他列
    )

# 插入行添加一个新的标识列
inserted_df = insert_rows.withColumn("inserted_column", lit("insert_value"))

# 合并目标Dataframe和插入行
df2 = df2.union(inserted_df.select(df2.columns))

# 保存到文件
df2.write.option("header", "true").csv("path/to/output.csv")

# 写回数据库（示例为MySQL）
df2.write \
    .format("jdbc") \
    .option("url", "jdbc:mysql://localhost/mydatabase") \
    .option("dbtable", "mytable") \
    .option("user", "myusername") \
    .option("password", "mypassword") \
    .mode("overwrite") \
    .save()