pyspark 新增数据 并更新旧数据

在 PySpark 中，可以使用 Delta Lake 来实现新增数据并更新旧数据。Delta Lake 是一个开源的数据湖解决方案，它提供了 ACID 事务支持、版本管理、数据合并等功能，可以用于构建企业级的数据湖。下面是一个简单的 PySpark 新增数据并更新旧数据示例：

from pyspark.sql import SparkSession

# 创建 SparkSession
spark = SparkSession.builder \
    .appName("Incremental Data Writing with Delta Lake") \
    .getOrCreate()

# 读取已有的 Delta Lake 表
df = spark.read.format("delta").load("/path/to/delta-table")

# 新增数据
new_data = spark.createDataFrame([(4, "D")], ["id", "value"])

# 将新增数据写入 Delta Lake 表，如果 "id" 已存在，则更新 "value"
df.union(new_data).write \
    .format("delta") \
    .mode("overwrite") \
    .option("mergeSchema", "true") \
    .option("mergeIntoMatched", "true") \
    .option("mergeKey", "id") \
    .save("/path/to/delta-table")

# 关闭 SparkSession
spark.stop()

在上面的示例中，我们首先使用 `spark.read.format("delta").load("/path/to/delta-table")` 读取已有的 Delta Lake 表，然后使用 `df.union(new_data).write` 将新增的数据合并到原有的表中，并使用 `mergeKey` 指定合并的键。在写入数据时，需要使用 `mode("overwrite")` 指定覆盖模式，使用 `option("mergeSchema", "true")` 和 `option("mergeIntoMatched", "true")` 开启合并模式。这样，如果 "id" 已经存在，就会更新 "value" 字段。注意，这里覆盖模式会删除原有的表并重新创建表，因此需要谨慎使用。