将数据集写入 MS SQL 服务器失败

Posted 2023-03-27

【中文标题】将数据集写入 MS SQL 服务器失败

【英文标题】：Writing Dataset to MS SQL server fails

【发布时间】：2019-12-23 17:44:01

【问题描述】：

java 8 mssql-jdbc: 7.4.0.jre8 spark-core_2.12：2.4.4 spark-sql_2.12：2.4.4

我正在尝试获取来自 MS SQL 2007 Server 表的 spark 数据集，并在经过几次转换后将其写入同一数据库上的另一个现有表（也因新表名而失败）。数据集的内容在执行 .show() 时似乎很好。无论我尝试使用 SaveMode.Append 还是 SaveMode.Overwrite 编写，spark 都会尝试创建一个新表并失败并出现错误：com.microsoft.sqlserver.jdbc.SQLServerException: Incorrect syntax near ';'。在调试时，jdbc 驱动程序会中断以下语句：

CREATE TABLE "RMS.cacr_rms.POC_REFUND_DETAILS"; ("REFUND_ID" NVARCHAR(MAX) , "ORDER_ID" NVARCHAR(MAX) , "PROFILE_ID" NVARCHAR(MAX) , "AGENT_ID" INTEGER , "CONSUMER_ID" INTEGER , "REASON_CODE_ID" INTEGER , "REFUND_TYPE" INTEGER , "STORE_ID" INTEGER , "STATUS" NVARCHAR(MAX) , "REFUND_CREATED_DATE" DATETIME , "LAST_UPDATED_DATE" INTEGER , "ORDER_SUBMITTED_DATE" DATETIME , "SHIP_DATE" INTEGER , "ORDER_AMOUNT" INTEGER , "REFUND_AMOUNT" INTEGER , "REFUND_CHANNEL" INTEGER , "TENANT" INTEGER , "VERTICAL" INTEGER , "ATG_REFUND_ID" NVARCHAR(MAX) )

附加信息：spark sql 中的 JdbcRelationProvider 类具有属性 tableExists false。

完整的堆栈跟踪：

    com.microsoft.sqlserver.jdbc.SQLServerException: Incorrect syntax near ';'.
    at com.microsoft.sqlserver.jdbc.SQLServerException.makeFromDatabaseError(SQLServerException.java:262)
    at com.microsoft.sqlserver.jdbc.SQLServerStatement.getNextResult(SQLServerStatement.java:1624)
    at com.microsoft.sqlserver.jdbc.SQLServerStatement.doExecuteStatement(SQLServerStatement.java:868)
    at com.microsoft.sqlserver.jdbc.SQLServerStatement$StmtExecCmd.doExecute(SQLServerStatement.java:768)
    at com.microsoft.sqlserver.jdbc.TDSCommand.execute(IOBuffer.java:7194)
    at com.microsoft.sqlserver.jdbc.SQLServerConnection.executeCommand(SQLServerConnection.java:2979)
    at com.microsoft.sqlserver.jdbc.SQLServerStatement.executeCommand(SQLServerStatement.java:248)
    at com.microsoft.sqlserver.jdbc.SQLServerStatement.executeStatement(SQLServerStatement.java:223)
    at com.microsoft.sqlserver.jdbc.SQLServerStatement.executeUpdate(SQLServerStatement.java:711)
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.createTable(JdbcUtils.scala:859)
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:81)
    at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:46)
    at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:70)
    at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:68)
    at org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(commands.scala:86)
    at org.apache.spark.sql.execution.SparkPlan.$anonfun$execute$1(SparkPlan.scala:131)
    at org.apache.spark.sql.execution.SparkPlan.$anonfun$executeQuery$1(SparkPlan.scala:155)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
    at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127)
    at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:80)
    at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:80)
    at org.apache.spark.sql.DataFrameWriter.$anonfun$runCommand$1(DataFrameWriter.scala:676)
    at org.apache.spark.sql.execution.SQLExecution$.$anonfun$withNewExecutionId$1(SQLExecution.scala:78)
    at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:125)
    at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:73)
    at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:676)
    at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:290)
    at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:271)
    at RmsSparkSession.write(RmsSparkSession.java:27)
    at MainApp.main(MainApp.java:28)

触发错误的代码：

ds.write().mode(SaveMode.Append).format("jdbc").option("url", config.getString("sqlserver.url"))
        .option("dbtable", tableName).option("user", config.getString("sqlserver.username"))
        .option("password", config.getString("sqlserver.password")).save();```

【问题讨论】：

你确定你需要那个分号吗？在定义其字段之前，创建表不是一个完整的语句吗？我在这里猜测——试图提供帮助

Jeremy 是正确的 - 分号是错误的。但是我看到尝试的表模式存在各种其他问题，并且会猜测导致此错误的代码还有其他问题。非常怀疑您的大多数（可能是全部） nvarchar(max) 列实际上应该是字符串。它们也不应该是巨大的字符串。 Last_Updated_Date（和其他）不应该是整数。您的表名会使任何试图使用它的人感到困惑，因为您使用 3 部分字符串作为表名。金额通常不是整数。

@Smor 读取数据时spark隐式定义schema。我可以自己显式定义一个模式，但这里的问题似乎是 spark 找不到表并且也无法创建新表，因为生成的 sql 中存在语法错误

【参考方案1】：

SaveMode.Overwrite 将创建表，SaveMode.Append 将追加到现有表而不创建它。

请试试这个语法，它对我有用。

    Properties connectionProps = new Properties();
    connectionProps.put("user", config.getString("sqlserver.username"));
    connectionProps.put("password", config.getString("sqlserver.password");
    connectionProps.put("driver", connection details);

    dataframe.write.mode(SaveMode.Append).jdbc(jdbcURL, tableName, connectionProps)

有关设置 JDBC 属性的信息，请参阅 https://docs.oracle.com/javase/tutorial/jdbc/basics/connecting.html。

【讨论】：

对我来说仍然失败。即使附加保存模式，Spark 也会尝试使用错误的 SQL 创建表。起作用的是使用 azure-sqldb-spark 连接器并进行批量复制，因为无论如何我都想优先考虑速度。

我不知道为什么，上面的代码在 Oracle 上以追加和覆盖模式都有效。如果有错误，您是否尝试过充电火花版本。我用过 2.4.2

不幸的是，我无法让它与 MS SQL 服务器一起工作。但是，使用 azure-sqldb-spark 连接器适用于标准 jdbc 写入和批量插入写入。连接器在内部使用相同的 jdbc 驱动程序。在处理大数据时，批量插入模式似乎也提供了显着的性能改进。