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" 字段进行合并
df.union(new_data).write \
    .format("delta") \
    .mode("append") \
    .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` 指定合并的键。最后，使用 `save` 将合并后的结果写入 Delta Lake 表中。在写入数据时，需要使用 `mode("append")` 指定追加模式，使用 `option("mergeSchema", "true")` 和 `option("mergeIntoMatched", "true")` 开启合并模式。