在Spark中实现SCD类型2

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, lit, when, current_timestamp

# 创建SparkSession
spark = SparkSession.builder.appName("SCD Type 2").getOrCreate()

# 读取源表数据
source_df = spark.read.option("header", "true").csv("source_table.csv")

# 处理维度数据变化
processed_df = source_df.withColumn("valid_to", lit(None)).withColumn("valid_from", current_timestamp())

# 识别新记录和更新记录
new_records_df = processed_df.filter(col("valid_to").isNull())
update_records_df = processed_df.filter(~col("valid_to").isNull())

# 更新历史记录
update_records_df = update_records_df.withColumn("valid_to", current_timestamp())

# 合并结果
final_df = new_records_df.union(update_records_df)

# 将结果写入目标表
final_df.write.mode("overwrite").option("header", "true").csv("target_table.csv")