sql-server - PySpark 1.5 和 MSSQL jdbc

Question

我在 Cloudera YARN 上的 Spark 1.5 上使用 PySpark，在 Centos 6 机器上使用 Python 3.3。 SQL Server 实例是 SQL Server Enterprise 64 位。下面列出了 SQL Server 驱动程序； sqljdbc4.jar；我已经添加到我的 .bashrc

export SPARK_CLASSPATH="/var/lib/spark/sqljdbc4.jar"
export PYSPARK_SUBMIT_ARGS="--conf spark.executor.extraClassPath="/var/lib/spark/sqljdbc4.jar" --driver-class-path="/var/lib/spark/sqljdbc4.jar" --jars="/var/lib/spark/sqljdbc4.jar" --master yarn --deploy-mode client"

当我启动 Spark 时，我可以看到确认

SPARK_CLASSPATH was detected (set to '/var/lib/spark/sqljdbc4.jar')

我有一个类似于此架构的数据框

root
 |-- daytetime: timestamp (nullable = true)
 |-- ip: string (nullable = true)
 |-- tech: string (nullable = true)
 |-- th: string (nullable = true)
 |-- car: string (nullable = true)
 |-- min_dayte: timestamp (nullable = true)
 |-- max_dayte: timestamp (nullable = true)

我已经在 MS SQL 服务器中创建了一个名为“dbo.shaping”的空表，其中 3 个时间戳列为 datetime2(7)，其他列为 nvarchar(50)。

我尝试使用此从 PySpark 导出数据帧

properties = {"user": "<username>", "password": "<password>"} 

df.write.format('jdbc').options(url='<IP>:1433/<dbname>', dbtable='dbo.shaping',driver="com.microsoft.sqlserver.jdbc.SQLServerDriver",properties=properties)

我收到以下回溯错误

Py4JError: An error occurred while calling o250.option. Trace:
py4j.Py4JException: Method option([class java.lang.String, class java.util.HashMap]) does not exist
at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:333)
at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:342)
at py4j.Gateway.invoke(Gateway.java:252)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:207)
at java.lang.Thread.run(Thread.java:744)

我的方法至少是正确的，也许这个错误与写入特定类型的数据有关，即我的数据构造有问题，而不是我的代码有问题？

Answer 1

您不能使用dict作为options的值。 options 方法仅需要 str 参数( Scala docs 和 PySpark annotations )，并扩展为单独调用 Java option。

在当前的 Spark 版本中，值为 automatically converted to string ，所以你的代码会默默地失败，但是 it isn't the case in 1.5 .

由于 properties 无论如何都是特定于 JDBC 驱动程序的，因此您应该使用 jdbc 方法:

properties = {
    "user": "<username>", "password": "<password>", "driver": 
    "com.microsoft.sqlserver.jdbc.SQLServerDriver"}

df.write.jdbc(
    url='<IP>:1433/<dbname>',
    table='dbo.shaping',
    properties=properties)

虽然解包属性也应该有效:

.options(
    url='<IP>:1433/<dbname>',
    dbtable='dbo.shaping',
    driver="com.microsoft.sqlserver.jdbc.SQLServerDriver",
    **properties)

一般来说，当您看到:

py4j.Py4JException: Method ... does not exist

它通常表示本地 Python 类型与所使用的 JVM 方法期望的类型之间不匹配。

另请参阅:How to use JDBC source to write and read data in (Py)Spark?