将数据从spark dataframe插入SQL server中的表

Question

我在数据库上使用Scala笔记本。我需要执行一个INSERT的数据从一个数据帧到一个表SQL服务器。如果数据已经存在，则不需要修改或插入只插入不存在的数据。

我尝试了这里指定的方法https://docs.databricks.com/spark/latest/data-sources/sql-databases.html#write-然而，它们并没有解决我的用例。保存模式。Append创建重复的数据项SaveMode。覆盖将替换现有数据（表），即SaveMode。如果表已经存在，“忽略”不会添加任何新数据。

df.write.mode（SaveMode.jdbc（url=dbUrl， table=table_name， dbConnectionProperties）

如何仅对数据库执行新数据的INSERT？

提前感谢您的帮助！

Answer 1

为了避免引入整个集合来进行比较，可以在SQL表上放置一个唯一的索引，并使用忽略重复项的选项。

MS文档：创建唯一索引

CREATE [ UNIQUE ] [ CLUSTERED | NONCLUSTERED ] INDEX index_name
    ON <object> ( column [ ASC | DESC ] [ ,...n ] )
    [ INCLUDE ( column_name [ ,...n ] ) ]
    [ WHERE <filter_predicate> ]
    [ WITH ( <relational_index_option> [ ,...n ] ) ]
    [ ON { partition_scheme_name ( column_name )
         | filegroup_name
         | default
         }
    ]
    [ FILESTREAM_ON { filestream_filegroup_name | partition_scheme_name | "NULL" } ]

[ ; ]

<object> ::=
{ database_name.schema_name.table_or_view_name | schema_name.table_or_view_name | table_or_view_name }

<relational_index_option> ::=
{
  | IGNORE_DUP_KEY = { ON | OFF }
}

Answer 2

注意：-这是一个解决办法。这不是一个全面的解决方案。

可以有解决此问题的变通方法。您需要在SQL服务器表中维护一个自动增量键/主键。源数据也应该在插入前的数据中有这个键。

可能出现以下情况：

New Primary key == Old Primary key -> job will fail with constraints exception.
New Primary key != Old Primary key -> insert successfully.

插入表失败可以在程序级别处理。

Answer 3

假设当前数据帧为df1。

您应该将SQL表中的现有数据读入另一个数据帧（df2）。然后使用减法（或减法键）：http://spark.apache.org/docs/latest/api/python/pyspark.html?highlight=subtract

val dfFinal = df1.subtract(df2)

dfFinal将包含要插入的剩余记录。