在数据砖自动加载器中处理重复项

Question

我是这个Databricks自动加载器的新手，我们有一个要求，我们需要通过Databricks自动加载器处理从AWS s3到delta表的数据。我正在测试这个自动加载程序，所以我遇到了重复的问题，即如果我上传一个文件名为emp_09282021.csv的文件，该文件与emp_09272021.csv文件具有相同的数据，那么它没有检测到任何重复，它只是简单地插入它们，所以如果我在emp_09272021.csv文件中有5行，现在当我上传emp_09282021.csv文件时，它将变成10行。

下面是我尝试的代码：

spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "csv") \
  .option("header",True) \
  .schema("id string,name string, age string,city string") \
  .load("s3://some-s3-path/source/") \
  .writeStream.format("delta") \
  .option("mergeSchema", "true") \
  .option("checkpointLocation", "s3://some-s3-path/tgt_checkpoint_0928/") \
  .start("s3://some-s3-path/spark_stream_processing/target/")

请指导我如何处理这件事？

Answer 1

自动加载器的任务不是检测副本，它为您提供摄取数据的可能性，但您需要自己处理副本。对此有几种方法:

< li >使用内置的< code>dropDuplicates函数。建议将它与水印一起使用，以避免创建一个巨大的状态，但您需要有一些列将被用作事件时间，并且它应该是< code>dropDuplicate列表的一部分(有关更多详细信息，请参见文档):

streamingDf \
  .withWatermark("eventTime", "10 seconds") \
  .dropDuplicates("col1", "eventTime")

• 使用Delta的合并功能-您只需要插入Delta表中没有的数据，但需要使用foreachBatch。类似这样的内容（请注意，表应该已经存在，或者您需要添加对不存在表的处理）：

from delta.tables import *

def drop_duplicates(df, epoch):
  table = DeltaTable.forPath(spark, 
      "s3://some-s3-path/spark_stream_processing/target/")
   dname = "destination"
   uname = "updates"
   dup_columns = ["col1", "col2"]
   merge_condition = " AND ".join([f"{dname}.{col} = {uname}.{col}"
      for col in dup_columns])
   table.alias(dname).merge(df.alias(uname), merge_condition)\
     .whenNotMatchedInsertAll().execute()

# ....
spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "csv") \
  .option("header",True) \
  .schema("id string,name string, age string,city string") \
  .load("s3://some-s3-path/source/") \
  .writeStream.foreachBatch(drop_duplicates)\
  .option("checkpointLocation", "s3://some-s3-path/tgt_checkpoint_0928/") \
  .start()

在这段代码中，您需要更改< code>dup_columns变量来指定用于检测重复项的列。