使用"INSERT OVERWRITE“使用databricks增量捕获更改数据

from pyspark.sql import SparkSession
from delta import configure_spark_with_delta_pip

# 配置 Spark 会话
spark = SparkSession.builder \
    .appName("Delta Lake Incremental Update") \
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
    .getOrCreate()

# 创建 Delta 表
data = [(1, "Alice"), (2, "Bob"), (3, "Charlie")]
columns = ["id", "name"]
df = spark.createDataFrame(data, columns)
df.write.format("delta").save("/delta/table")

# 插入新数据
new_data = [(4, "David"), (5, "Eve")]
new_df = spark.createDataFrame(new_data, columns)
new_df.write.mode("overwrite").format("delta").save("/delta/table")

# 增量更新数据
update_data = [(1, "Alicia"), (2, "Bobby")]
update_df = spark.createDataFrame(update_data, columns)
update_df.write.mode("overwrite").format("delta").option("mergeStrategy", "replaceWhere").save("/delta/table")

# 查询表数据
result_df = spark.read.format("delta").load("/delta/table")
result_df.show()