使用 JDBC 从 Pyspark 更新表答案

【问题标题】：update table from Pyspark using JDBC使用 JDBC 从 Pyspark 更新表
【发布时间】：2020-04-24 00:15:27
【问题描述】：

我有一个小的 log 数据框，其中包含有关在给定笔记本中执行的 ETL 的元数据，该笔记本是在 Azure DataFactory 中管理的更大的 ETL 管道的一部分。

不幸的是，Databricks 似乎无法调用存储过程，因此我手动将包含正确数据的行附加到我的日志表中。

但是，在给定一组条件的情况下，我无法找出正确的 sytnax 来更新表：

我用来追加单行的语句如下：

spark_log.write.jdbc(sql_url, 'internal.Job',mode='append')

然而，这可以正常工作，因为我的数据工厂正在调用存储过程，

我需要处理像

这样的查询

query  = f"""
UPDATE [internal].[Job] SET
     [MaxIngestionDate]                date                   {date}
,    [DataLakeMetadataRaw]            varchar(MAX)            NULL
,    [DataLakeMetadataCurated]        varchar(MAX)            NULL
WHERE [IsRunning] = 1
AND [FinishDateTime] IS NULL"""

这可能吗？如果是这样，有人可以告诉我怎么做吗？

查看文档，这似乎只提到了使用带有query 参数的select 语句：

目标数据库是 Azure SQL 数据库。

https://spark.apache.org/docs/latest/sql-data-sources-jdbc.html

只是添加这是一个很小的操作，因此性能不是问题。

【问题讨论】：

对于任何在这里想知道的迷失的灵魂，我的工作是在我的数据工厂管道中完成笔记本时传递一个 json blob，然后我将其解析并作为参数传递给我的存储过程，这反过来更新了我的日志表。

标签： azure pyspark databricks

【解决方案1】：

您不能在 Spark 中使用带有数据帧的 jdbc 进行单条记录更新。您只能追加或替换整个表。

您可以使用 pyodbc 进行更新 - 需要安装 MSSQL ODBC 驱动程序 (How to install PYODBC in Databricks) 或者您可以通过 JayDeBeApi (https://pypi.org/project/JayDeBeApi/) 使用 jdbc

【讨论】：