从 Azure Synapse 中的 Apache Spark 将数据写入 SQL DW

Posted 2023-03-25

【中文标题】从 Azure Synapse 中的 Apache Spark 将数据写入 SQL DW

【英文标题】：Write Data to SQL DW from Apache Spark in Azure Synapse

【发布时间】：2021-08-26 16:25:26

【问题描述】：

当我从 Databricks 将数据写入 Azure 中的 SQL DW 时，我使用以下代码：

example1.write.format("com.databricks.spark.sqldw").option("url", sqlDwUrlSmall).option("dbtable", "SampleTable12").option("forward_spark_azure_storage_credentials","True") .option("tempdir", tempDir).mode("overwrite").save()

这不适用于 Synapse Notebook 中的 Notebook。我得到了错误：

Py4JJavaError: An error occurred while calling o174.save.
: java.lang.ClassNotFoundException: Failed to find data source: com.databricks.spark.sqldw. Please find packages at http://spark.apache.org/third-party-projects.html
    at org.apache.spark.sql.execution.datasources.DataSource$.lookupDataSource(DataSource.scala:656) Caused by: java.lang.ClassNotFoundException: com.databricks.spark.sqldw.DefaultSource

基本上，对于 Azure Synapse 中的 Apache Spark，我需要知道 com.databricks.spark.sqldw 的等价物。

谢谢

【参考方案1】：

如果您要写入与笔记本相同的 Synapse 工作区中的专用 SQL 池，那么它就像调用 synapsesql 方法一样简单。 Scala 中的一个简单参数化示例，使用 Synapse 笔记本的参数单元功能。

// Read the table
val df = spark.read.synapsesql(s"$pDatabaseName.$pSchemaName.$pTableName")

// do some processing ...

// Write it back with _processed suffixed to the table name
df.write.synapsesql(s"$pDatabaseName.$pSchemaName.$pTableName_processed", Constants.INTERNAL)

如果您尝试从笔记本写入不同的专用 SQL 池或旧的 Azure SQL 数据仓库，那么情况会有所不同，但有一些很好的示例 here。

更新：带有美元符号的大括号中的项目（例如$pDatabaseName）是参数。您可以在笔记本中指定一个参数单元，以便可以从外部传递参数，例如使用执行笔记本活动从 Azure 数据工厂 (ADF) 或 Synapse 管道，并在笔记本中重复使用，如我上面的示例所示。详细了解 Synapse Notebook 参数here。

【讨论】：

感谢您在这方面的坚持。我添加了以下内容，但我收到错误“不完整的声明”%%spark val pDatabaseName = "MyFirstSQLPool" val pSchemaName = "dbo" val pTableName = "mysampletable3" val df.write.synapsesql(s"$pDatabaseName.$pSchemaName.$pTableName_processed", Constants.INTERNAL)

我的做法是有两个单元格，单元格 1 有三个参数，并确保该单元格被指定为参数单元格。您可以通过单击单元格的右上角来执行此操作。然后让 Cell 2 使用这些参数。有意义吗？

另外，代码中填充数据框 (df) 的元素在哪里？

早上@wBob。我不完全确定你的意思。填充 df 的代码是 test = spark.sql("select * from testtable")

您原来的解决方案就像做梦一样。问题是 SQL 专用池已暂停.. doh！