如何在 pyspark 中使用 azure-sqldb-spark 连接器

Question

我想每天使用 PySpark 将大约 10 GB 的数据写入 Azure SQL 服务器数据库。目前使用的是 JDBC 驱动程序，这需要花费数小时来逐个插入语句。

我打算使用 azure-sqldb-spark 连接器，它声称使用批量插入来加速写入。

我浏览了官方文档：https://github.com/Azure/azure-sqldb-spark。 该库是用 scala 编写的，基本上需要使用 2 个 scala 类：

import com.microsoft.azure.sqldb.spark.config.Config
import com.microsoft.azure.sqldb.spark.connect._

val bulkCopyConfig = Config(Map(
  "url"               -> "mysqlserver.database.windows.net",
  "databaseName"      -> "MyDatabase",
  "user"              -> "username",
  "password"          -> "*********",
  "databaseName"      -> "MyDatabase",
  "dbTable"           -> "dbo.Clients",
  "bulkCopyBatchSize" -> "2500",
  "bulkCopyTableLock" -> "true",
  "bulkCopyTimeout"   -> "600"
))

df.bulkCopyToSqlDB(bulkCopyConfig)

是否可以像这样在pyspark中使用（使用sc._jvm）：

Config = sc._jvm.com.microsoft.azure.sqldb.spark.config.Config
connect= sc._jvm.com.microsoft.azure.sqldb.spark.connect._

//all config

df.connect.bulkCopyToSqlDB(bulkCopyConfig)

我不是 Python 方面的专家。任何人都可以帮助我完成完整的 sn-p 以完成这项工作。

Answer 1

目前（截至 2019 年 3 月）Spark 连接器仅支持 Scala API（如文档中的 here 所述）。 因此，如果您在笔记本上工作，则可以在 python 中进行所有预处理，最后将数据帧注册为临时表，例如。 G。 ：

df.createOrReplaceTempView('testbulk')

并且必须在 Scala 中完成最后一步：

%scala
//configs...
spark.table("testbulk").bulkCopyToSqlDB(bulkCopyConfig)