Py4JJavaError java.lang.NullPointerException org.apache.spark.sql.DataFrameWriter.jdbc

Question

Ich habe diesen Fehler, wenn ich versuchte, ein Spark-Datenframe zu Postgres DB zu schreiben. Ich bin mit einem lokalen Cluster und der Code ist wie folgt:

from pyspark import SparkContext 
from pyspark import SQLContext, SparkConf 
import os 

os.environ["SPARK_CLASSPATH"] = '/usr/share/java/postgresql-jdbc4.jar' 

conf = SparkConf() \ 
.setMaster('local[2]') \ 
.setAppName("test") 

sc = SparkContext(conf=conf) 
sqlContext = SQLContext(sc) 

df = sc.parallelize([("a", "b", "c", "d")]).toDF() 

url_connect = "jdbc:postgresql://localhost:5432" 
table = "table_test" 
mode = "overwrite" 
properties = {"user":"postgres", "password":"12345678"} 
df.write.option('driver', 'org.postgresql.Driver').jdbc(
    url_connect, table, mode, properties) 

Das Fehlerprotokoll ist wie folgt:

Py4JJavaError: An error occurred while calling o119.jdbc. 
: java.lang.NullPointerException 
at org.apache.spark.sql.DataFrameWriter.jdbc(DataFrameWriter.scala:308) 
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 
at java.lang.reflect.Method.invoke(Method.java:498) 
at py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:231) 
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:381) 
at py4j.Gateway.invoke(Gateway.java:259) 
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:133) 
at py4j.commands.CallCommand.execute(CallCommand.java:79) 
at py4j.GatewayConnection.run(GatewayConnection.java:209) 
at java.lang.Thread.run(Thread.java:745) 

ich versucht habe, eine Antwort von der Web-Suche, konnte aber keine finden. Vielen Dank im Voraus!

Answer 1

Haben Sie versucht, die Datenbank in Ihrer table_test Variable anzugeben? Ich habe eine ähnliche Implementierung, die wie folgt aussieht:

mysqlUrl = "jdbc:mysql://mysql:3306" 
properties = {'user':'root', 
       'password':'password', 
       'driver':'com.mysql.cj.jdbc.Driver' 
       } 
table = 'db_name.table_name' 

try: 
    schemaDF = spark.read.jdbc(mysqlUrl, table, properties=properties) 
    print 'schema DF loaded' 
except Exception, e: 
    print 'schema DF does not exist!'