PySpark 1.5和MSSQL jdbc

Question

我在Cloudera YARN上的Spark 1.5上使用PySpark,在Centos 6 Machines上使用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

您不能将a dict用作值options.options方法只需要str参数(Scala docs和PySpark注释),并且扩展为单独调用Java option.

在当前的Spark版本中,值会自动转换为字符串,因此您的代码将以静默方式失败,但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方法所期望的类型之间不匹配.

另请参见:如何使用JDBC源在(Py)Spark中写入和读取数据？