Databricks - 写入 Azure Synapse 时出错

Posted 2023-03-23

【中文标题】Databricks - 写入 Azure Synapse 时出错

【英文标题】：Databricks - Error writing to Azure Synapse

【发布时间】：2021-07-16 04:32:08

【问题描述】：

我正在尝试使用以下代码将数据写入带有标识字段的 Azure Synapse 表

数据块上的代码

def get_jdbc_connection(host, sqlDatabase, user, password):
  jdbcHostname = ".database.windows.net".format(host)
  jdbc_url = "jdbc:sqlserver://:1433;database=;user=@;password=;encrypt=true;trustServerCertificate=false;hostNameInCertificate=*.database.windows.net;loginTimeout=30;".format(jdbcHostname, sqlDatabase, user, host, password)
  url = "jdbc:sqlserver://:1433;database=;encrypt=true;trustServerCertificate=false;hostNameInCertificate=*.database.windows.net;loginTimeout=30;".format(jdbcHostname, sqlDatabase)
  return (jdbc_url,url )

def write_adw(spark, df_target_adw, jdbc_url, table, tempDir, option_mode, pre_Actions ):
    df_target_adw.write.format("com.databricks.spark.sqldw") \
        .option("url", jdbc_url) \
        .option("useAzureMSI", "true") \
        .option("preActions", pre_Actions) \
        .option("dbTable", table) \
        .option("tempDir", tempDir) \
        .mode(option_mode) \
        .save()
    
dftraffic = spark.sql('SELECT distinct SourceName\
              ,1 AS IsActiveRow \
             ,"Pipe-123" as pipelineId \
              ,current_timestamp as ADFCreatedDateTime \
              ,current_timestamp as ADFModifiedDateTime \
              from deltaTable')

#write to ADW
(jdbc_url, url_adw) = get_jdbc_connection(host, sqlDatawarehouse,user, password)
target_table = 'TargetTable_name'
option_mode= "append"
pre_Actions= " SELECT GETDATE()"
write_adw(spark, dftraffic, jdbc_url, target_table, tempDir, option_mode, pre_Actions )

adw 上的目标表架构

Column Name	Data Type
SourceSID	INT IDENTITY (1,1) NOT NULL
Source Name	VARCHAR(20) NOT NULL
IsRowActive	BIT NOT NULL
PipelineId	VARCHAR(20) NOT NULL
ADFCreatedDateTime	DATETIME NOT NULL
ADFModifiedDateTime	DATETIME NOT NULL

databricks 上的配置详情

Databricks 运行时 7.4（包括 Apache Spark 3.0.1、Scala 2.12）

错误信息

Py4JJavaError：调用 o457.save 时出错。 ：com.databricks.spark.sqldw.SqlDWSideException：Azure Synapse Analytics 无法执行连接器生成的 JDBC 查询。 基础 SQLException(s): - com.microsoft.sqlserver.jdbc.SQLServerException: 表中标识列的显式值只能在使用列列表并且 IDENTITY_INSERT 为 ON 时指定

代码在 databricks 运行时 6.4 Spark 2.4.5 上运行良好，我在尝试升级 dbk 运行时时遇到了这个错误。 我怎样才能让它工作？

【问题讨论】：

您是否尝试将SET IDENTITY_INSERT archive_table ON; 添加为pre_Actions

让我试试，然后更新你。

@Idleguys 你有解决问题的办法吗？

不，这不起作用，因为我需要在插入完成后“关闭”桌子上的 identity_insert。

【参考方案1】：

你没有额外的行“1 AS IsActiveRow”吗？我在架构中没有看到

dftraffic = spark.sql('SELECT distinct SourceName\
              ,1 AS IsActiveRow \
             ,"Pipe-123" as pipelineId \
              ,current_timestamp as ADFCreatedDateTime \
              ,current_timestamp as ADFModifiedDateTime \
              from deltaTable)

【讨论】：

感谢您的回复。上述架构定义中有错字。道歉！我的目标表上有“IsActiveRow”字段。