PySpark：以“追加”模式写入并在某些条件匹配时覆盖

Question

我将以下 Spark 数据帧附加到现有的 Redshift 数据库。如果month = '2021-12' and state = 'ga'

，我想使用“月”和“州”作为检查和替换 Redshift 表中数据的标准

我们要附加的数据框：df

month	state	product
2021-12	ca	ho0
2021-12	ca	ho1
2021-12	ca	ho2
2021-12	ca	ho3
2021-12	ca	ho4
2021-12	ga	ho5
2021-12	ga	ho6
2021-12	ga	ho7
2021-12	ga	ho8
2021-12	ga	ho9

我尝试了以下脚本来附加它，看起来它只附加 df 数据帧，而不替换（覆盖）“2021-12”月和状态“ga”的现有记录。

df.write \
  .format("xxx") \
  .option("url", "xxx") \
  .option("dbtable", "table1") \
  .option("tempdir", "xxxx") \
  .option("aws_iam_role", "xxxx") \
  .mode("append") \
  .option("replaceWhere", "month == '2021-12' AND state == 'ga'") \
  .save()

我猜.option("replaceWhere", "month == '2021-12' AND state == 'ga'") 不工作。我怎样才能做出改变？谢谢！ （我也试过下面的部分，好像现有的记录都没有了，换成df）

  .mode("overwrite") \
  .option("replaceWhere", "month == '2021-12' AND state == 'ga'") \

Answer 1

replaceWhere 这个选项几乎就像一个动态覆盖分区，基本上你告诉 Spark 只覆盖那些范围分区上的数据。另外，只有当你的dataframe符合条件replaceWhere时才会保存数据，否则如果单行不匹配，则会抛出异常Data write out does not match replaceWhere。你是参考link .如果你想在分区内处理相同的场景，你可以使用MERGE INTO。

Answer 2

Spark 不能直接与 Redshift 进行数据合并，因为我们使用的是 Databricks，它提供了“Delta Lake”作为进行数据更新/合并的环境，然后我们将表从 Delta Lake 写入 Redshift。 https://docs.databricks.com/delta/delta-update.html#language-python

Answer 3

使用 Redshift 时，您需要点击以下链接。比您需要的多一点，但适用。该链接显示 UPSERT 模拟。链接是 http://www.silota.com/blog/amazon-redshift-upsert-support-staging-table-replace-rows/

即使插入到 Redshift 也不是 Spark 的事情，我过去曾在此发布过。一个标准使用 s3 中的 COPY。