spark-redshift-connector：将保存到 redshift 与删除查询相结合答案

【问题标题】：spark-redshift-connector: combine saving to redshift with a delete queryspark-redshift-connector：将保存到 redshift 与删除查询相结合
【发布时间】：2016-11-20 05:21:15
【问题描述】：

我想使用 spark-redshift-connector (scala) 定期更新 Redshift。每次更新都以删除操作开始。（我正在向 Redshift 执行一种 Upsert）

有没有办法可以用库来执行它？可以用交易吗？

任何建议将不胜感激。

谢谢，伊兰。

【问题讨论】：

标签： apache-spark amazon-redshift

【解决方案1】：

似乎有一个名为“preactions”（和“postactions”）的选项可以解决问题。

查看文档： https://github.com/databricks/spark-redshift

【讨论】：

您能分享一个工作示例以供参考吗？我是否必须无论如何将火花数据帧发送到 Redshift 的着陆区才能有条件地删除 upsert？说： ----------- ``` 使用 test_1 从 test_2 中删除，其中 test_2.a=test_1.a 和 test_2.b=test1.b``` ----- 我们必须有redshift 中的 test_1 和 test_2 ---- pre-actions 在这里有什么帮助？

【解决方案2】：

请参考以下示例：

val min_date=mydf.select(min("actual_ship_date")).rdd.map(line=>line(0)).take(1)
val max_date=mydf.select(max("actual_ship_date")).rdd.map(line=>line(0)).take(1)
val query="delete from semi_sdt.kgd_tsb_shippment where 
actual_ship_date>='"+min_date(0).toString+"' and 
actual_ship_date<='"+max_date(0).toString+"'"
//Write data to RedShift
mydf.coalesce(1).write.
format("com.databricks.spark.redshift").
option("url",redShiftUrl).
option("dbtable","semi_sdt.kgd_tsb_shippment").
option("tempdir",s3dir).
option("forward_spark_s3_credentials",true).
option("preactions",query).
mode("append").
save()

【讨论】：