PySpark 1.5 & MSSQL jdbc

Я использую PySpark на Spark 1.5 на Cloudera YARN, используя Python 3.3 на Centos 6 Machines. Экземпляр 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')

У меня есть dataframe, который выглядит как эта схема

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", где три столбца временной метки будут datetime2 (7), а остальные nvarchar (50).

Я пытаюсь экспортировать dataframe из 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)

Являются ли мои методы хотя бы правильными, и, возможно, эта ошибка связана с записью определенного типа данных, т.е. у меня проблема с конструкцией данных, а не с моим кодом?

Ответы

Ответ 1

Вы не можете использовать 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?