SQL 错误：“嵌套的 INSERT、UPDATE、DELETE 或 MERGE 语句必须具有 OUTPUT 子句。” - 在 Azure Databricks 中执行时

Posted 2023-04-15

【中文标题】SQL 错误：“嵌套的 INSERT、UPDATE、DELETE 或 MERGE 语句必须具有 OUTPUT 子句。” - 在 Azure Databricks 中执行时

【英文标题】：SQL Error: "A nested INSERT, UPDATE, DELETE, or MERGE statement must have an OUTPUT clause." - while executing in Azure Databricks

【发布时间】：2021-08-10 14:00:41

【问题描述】：

SQL 表 - TABLE1

Table1 包含三列 - STORE_ID (PK,numeric(5,0), not null), ATTRB_NAME (PK,varchar(50), not null), PRDCT_NAME (varchar(50), not null)。

STORE_ID 值变化，ATTRB_NAME、PRDCT_NAME 值不变

Azure databricks notebook 已开发用于执行插入查询，该查询实际上将数据插入到上表中。商店 id 取自一个列表。以下代码已执行：

from pyspark.sql import functions as F

insertDataList = []
store_id_list = [210, 345]

for element in store_id_list:
  paramList = ['ATTRB_VAL','PRDCT_VAL']
  paramList.insert(0,element)
  insertDataList.append(tuple(paramList))

rows = insertDataList
values = ', '.join(map(str, rows))
insert_query = "(INSERT INTO [DB1].[dbo].[TABLE1] VALUES ) b_alias".format(values)

jdbcUrl = "jdbc:sqlserver://0:1;database=2".format(jdbcHostname, jdbcPort, jdbcDatabase)
connectionProperties = 
  "user" : jdbcUsername,
  "password" : jdbcPassword,
  "driver" : "com.microsoft.sqlserver.jdbc.SQLServerDriver"

df = spark.read.jdbc(url=jdbcUrl, table=insert_query, properties=connectionProperties)

不幸的是，它抛出以下错误：

com.microsoft.sqlserver.jdbc.SQLServerException: 嵌套的 INSERT， UPDATE、DELETE 或 MERGE 语句必须有一个 OUTPUT 子句。

我通过重新构建插入查询来执行脚本，但最终都显示了相同的错误。我可以知道这里需要做什么吗？我是 SQL 的初学者。任何帮助将不胜感激。

【问题讨论】：

为什么你的 INSERT 声明在括号内 (())？另外，您为什么要注入这些值，而不是正确地参数化您的语句？

@Larnu microsoft docs 遵循该格式以使用来自 azure-databricks 的 python 执行查询。值正在注入，因为 store_id_list 将在外部提供。因此，必须首先从 store_id_list 和 paramList 创建一行，然后将其注入到插入查询中。 store_id_list 可以有 n 个元素。

该文档中的INSERT 语句均未在括号内...至于您注入的原因，您的评论暗示该表具有动态列数；这不可能。 参数化你的声明。

请帮助我重新构建它。当我执行不带括号的 INSERT 查询时，它给出了 Incorrect Syntax 错误。

删除括号和参数...INSERT INTO [dbo].[TABLE1] VALUES(?,?)（? 是参数的占位符，您需要按照pyspark 的语法传递）。我还删除了数据库名称，因为它应该在连接字符串中定义。

【参考方案1】：

感谢您的所有建议。试了下面的代码sn-p就可以成功插入数据了。

for rec in rows:
      df = spark.createDataFrame([(rec[0], rec[1], rec[2])],['STORE_ID', 'ATTRB_NAME', 'PRDCT_NAME']).cache()
      df.write.jdbc(url=jdbcUrl,table='TABLE1',mode='append',properties=connectionProperties)