我们如何在Apache Spark中实现CDC(变更数据捕获)?

首先,我假设你的PC上安装了Apache Spark和Python 3。 设置venv后,您需要安装 pyspark 使用pip命令打包

接下来,在使用get as spark会话后 SparkSession 。现在,是时候读取当前csv文件了

old_df = spark.read.csv("/data/2024-02-01.csv", header=True, inferSchema=True)
new_df = spark.read.csv("/data/2024-02-02.csv", header=True, inferSchema=True)

之后,您可能需要加入该数据帧。您可以根据要加入的列使用join函数。

最后打电话给 with列 使用difference参数的函数来获取您感兴趣的列中的更改。

希望这能有所帮助。如果您仍然收到任何错误,请分享错误。