删除功能与spark sql数据框架

Question

我需要为我的spark应用程序从postgres db加载/删除特定的记录。对于加载，我使用spark数据框架在以下格式

sqlContext.read.format("jdbc").options(Map("url" -> "postgres url", 
      "user" -> "user" ,
      "password" -> "xxxxxx" , 
      "table" -> "(select * from employee where emp_id > 1000) as filtered_emp")).load()

要删除数据，我直接写sql而不是使用dataframes

delete from employee where emp_id > 1000

问题是，是否有一个火花的方式删除数据库中的记录类似于下面的东西?或者唯一的方法是使用直接sql?

sqlContext.read.format("jdbc").options(Map("url" -> "postgres url", 
      "user" -> "user" ,
      "password" -> "xxxxxx" , 
      "table" -> "(delete from employee where emp_id > 1000) as filtered_emp")).load()

Answer 1

如果你想修改(删除记录)数据的实际来源，即表在postgres那么Spark不会是一个伟大的方式。您可以直接使用jdbc客户端来实现相同的目的。

如果你想这样做(基于你作为数据框架的一部分计算的一些线索的分布式方式);您可以将相同的JDBC客户端代码与具有用于删除记录的逻辑/触发器信息的数据框架对应编写，并且可以在多个worker上并行执行。

Answer 2

我有过类似的经历，在Scala中使用Java JDBC库从表中删除行。你可以在这里看到解决方案:使用Azure Databricks和Scala从Azure Sql表中删除行