将数据从 spark 数据帧插入到 SQL Server 中的表中答案

【问题标题】：INSERT data from spark dataframe to a table in SQL server将数据从 spark 数据帧插入到 SQL Server 中的表中
【发布时间】：2019-12-21 06:54:47
【问题描述】：

我在 Databricks 上使用 Scala Notebook。我需要将数据从数据框插入到 SQL Server 中的表中。如果数据已经存在，则无需修改或插入——只插入不存在的数据。

我尝试了这里指定的方法https://docs.databricks.com/spark/latest/data-sources/sql-databases.html#write-data-to-jdbc，但是，它们没有解决我的用例。 SaveMode.Append 创建数据的重复条目，SaveMode.Overwrite 替换现有数据（表），如果表已存在，SaveMode.Ignore 不会添加任何新数据。

df.write.mode(SaveMode.Overwrite).jdbc(url=dbUrl, table=table_name, dbConnectionProperties)

如何只向数据库插入新数据？

非常感谢您的帮助！

【问题讨论】：

标签： sql-server scala apache-spark-sql databricks

【解决方案1】：

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

MS Document: Create Unique Indexes

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 }
}

【讨论】：

【解决方案2】：

假设您当前的数据帧是 df1。

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

val dfFinal = df1.subtract(df2)

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

【讨论】：

谢谢！这种方法需要将大量数据（每个作业 5 亿+）引入 Databricks。是否建议在 spark 中而不是在 SQL 中进行重复检查和 INSERT？根据您的经验，它通常会提供更好的性能吗？我将对其进行测试，但只是想知道您是否已经对此进行了实验。
是的，这可能是个问题。理想情况下，您不会加载整个 500m。而是选择过去 24 小时内可能存在重复的记录。如果这不可能，另一种选择是将 df1 加载到临时表中，然后执行 SQL 语句以使用连接加载新记录。

【解决方案3】：

注意：- 这是一种解决方法。不是一个完整的证明解决方案。

这个问题可以有一个解决方法。您需要在 SQL Server 表中维护一个自增键/主键。并且源数据在插入之前也应该在数据中有这个键。

可能会出现以下情况：

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

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

【讨论】：

不是一个真正的答案。