10

How / where do I install the jdbc drivers for spark sql? I'm running the all-spark-notebook docker image, and am trying to pull some data directly from a sql database into spark.

From what I can tell I can tell I need to include the drivers in my Classpath, I'm just not sure how to do that from pyspark?

from pyspark.sql import SparkSession
spark = SparkSession \
    .builder \
    .master("local") \
    .appName("Python Spark SQL basic example") \
    .getOrCreate()

jdbcDF = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:postgresql:dbserver") \
    .option("dbtable", "jdbc:postgresql:dbserver") \
    .load()

---------------------------------------------------------------------------
Py4JJavaError                             Traceback (most recent call last)
<ipython-input-2-f3b08ff6d117> in <module>()
      2 spark = SparkSession     .builder     .master("local")     .appName("Python Spark SQL basic example")     .getOrCreate()
      3 
----> 4 jdbcDF = spark.read     .format("jdbc")     .option("url", "jdbc:postgresql:dbserver")     .option("dbtable", "jdbc:postgresql:dbserver")     .load()

/usr/local/spark/python/pyspark/sql/readwriter.py in load(self, path, format, schema, **options)
    163             return self._df(self._jreader.load(self._spark._sc._jvm.PythonUtils.toSeq(path)))
    164         else:
--> 165             return self._df(self._jreader.load())
    166 
    167     @since(1.4)

/usr/local/spark/python/lib/py4j-0.10.4-src.zip/py4j/java_gateway.py in __call__(self, *args)
   1131         answer = self.gateway_client.send_command(command)
   1132         return_value = get_return_value(
-> 1133             answer, self.gateway_client, self.target_id, self.name)
   1134 
   1135         for temp_arg in temp_args:

/usr/local/spark/python/pyspark/sql/utils.py in deco(*a, **kw)
     61     def deco(*a, **kw):
     62         try:
---> 63             return f(*a, **kw)
     64         except py4j.protocol.Py4JJavaError as e:
     65             s = e.java_exception.toString()

/usr/local/spark/python/lib/py4j-0.10.4-src.zip/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
    317                 raise Py4JJavaError(
    318                     "An error occurred while calling {0}{1}{2}.\n".
--> 319                     format(target_id, ".", name), value)
    320             else:
    321                 raise Py4JError(

Py4JJavaError: An error occurred while calling o36.load.
: java.sql.SQLException: No suitable driver
    at java.sql.DriverManager.getDriver(DriverManager.java:315)
    at org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions$$anonfun$7.apply(JDBCOptions.scala:84)
    at org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions$$anonfun$7.apply(JDBCOptions.scala:84)
    at scala.Option.getOrElse(Option.scala:121)
    at org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions.<init>(JDBCOptions.scala:83)
    at org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions.<init>(JDBCOptions.scala:34)
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:32)
    at org.apache.spark.sql.execution.datasources.DataSource.resolveRelation(DataSource.scala:306)
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:178)
    at org.apache.spark.sql.DataFrameReader.load(DataFrameReader.scala:146)
    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:244)
    at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
    at py4j.Gateway.invoke(Gateway.java:280)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:214)
    at java.lang.Thread.run(Thread.java:748)

2 Answers 2

9

In order to include the driver for postgresql you can do the following:

from pyspark.conf import SparkConf

conf = SparkConf()  # create the configuration
conf.set("spark.jars", "/path/to/postgresql-connector-java-someversion-bin.jar")  # set the spark.jars

...
spark = SparkSession.builder \
        .config(conf=conf) \  # feed it to the session here
        .master("local") \
        .appName("Python Spark SQL basic example") \
        .getOrCreate()

Now, since you are using Docker, I guess you have to mount the folder that has the driver jar and refer to the mounted folder. (e.g.: How to mount a host directory in a Docker container)

Hope this helps, good luck!

Edit: A diffferent way would be to give the --driver-class-path argument when using spark-submit like this:

spark-submit --driver-class-path=path/to/postgresql-connector-java-someversion-bin.jar file_to_run.py

but I'm guessing this is not how you will run this.

1
  • Does the conf.set() method set the spark.jars path permanently? I mean what is the difference between conf.set("spark.jars", "path/to/file.jar") and just doing.config("spark.jars", "path/to/file.jar")?
    – Banty
    Commented Jun 29, 2022 at 16:29
1

Putting the driver into the pyspark path works but the correct way to do it is to add something line this:

conf = pyspark.SparkConf().setAll([('spark.executor.id', 'driver'), 
                               ('spark.app.id', 'local-1631738601802'), 
                               ('spark.app.name', 'PySparkShell'), 
                               ('spark.driver.port', '32877'), 
                               ('spark.sql.warehouse.dir', 'file:/home/data_analysis_tool/spark-warehouse'), 
                               ('spark.driver.host', 'localhost'), 
                               ('spark.sql.catalogImplementation', 'hive'), 
                               ('spark.rdd.compress', 'True'), 
                               ('spark.driver.bindAddress', 'localhost'), 
                               ('spark.serializer.objectStreamReset', '100'), 
                               ('spark.master', 'local[*]'), 
                               ('spark.submit.pyFiles', ''), 
                               ('spark.app.startTime', '1631738600836'), 
                               ('spark.submit.deployMode', 'client'), 
                               ('spark.ui.showConsoleProgress', 'true'),
                               ('spark.driver.extraClassPath','/tmp/postgresql-42.2.23.jar')])

note the line:

('spark.driver.extraClassPath','/tmp/postgresql-42.2.23.jar')

Here is the whole code:

import psycopg2
import pandas as pd
import pyspark
from pyspark.sql import SparkSession
from sqlalchemy import create_engine
import qgrid

#appName = "PySpark PostgreSQL Example - via psycopg2"
#master = "local"

#spark = SparkSession.builder.master(master).appName(appName).getOrCreate()



conf = pyspark.SparkConf().setAll([('spark.executor.id', 'driver'), 
                                   ('spark.app.id', 'local-1631738601802'), 
                                   ('spark.app.name', 'PySparkShell'), 
                                   ('spark.driver.port', '32877'), 
                                   ('spark.sql.warehouse.dir', 'file:/home/data_analysis_tool/spark-warehouse'), 
                                   ('spark.driver.host', 'localhost'), 
                                   ('spark.sql.catalogImplementation', 'hive'), 
                                   ('spark.rdd.compress', 'True'), 
                                   ('spark.driver.bindAddress', 'localhost'), 
                                   ('spark.serializer.objectStreamReset', '100'), 
                                   ('spark.master', 'local[*]'), 
                                   ('spark.submit.pyFiles', ''), 
                                   ('spark.app.startTime', '1631738600836'), 
                                   ('spark.submit.deployMode', 'client'), 
                                   ('spark.ui.showConsoleProgress', 'true'),
                                   ('spark.driver.extraClassPath','/tmp/postgresql-42.2.23.jar')])


sc = pyspark.SparkContext(conf=conf)
sc.getConf().getAll()

sparkSession = SparkSession (sc)

sparkDataFrame = sparkSession.read.format("jdbc") \
    .options(
    url="jdbc:postgresql://localhost:5432/Database",
    dbtable="test_features_3",
    user="database_user",
    password="Pa$$word").load()

print (sparkDataFrame.count())
sc.stop()

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.