Pyspark connection to the Microsoft SQL server?

Question

I have a huge dataset in SQL server, I want to Connect the SQL server with python, then use pyspark to run the query.

I've seen the JDBC driver but I don't find the way to do it, I did it with PYODBC but not with a spark.

Any help would be appreciated.

Answer 1

Please use the following to connect to Microsoft SQL:

def connect_to_sql(
    spark, jdbc_hostname, jdbc_port, database, data_table, username, password
):
    jdbc_url = "jdbc:sqlserver://{0}:{1}/{2}".format(jdbc_hostname, jdbc_port, database)

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

    df = spark.read.jdbc(url=jdbc_url, table=data_table, properties=connection_details)
    return df

spark is a SparkSession object, and the rest are pretty clear.

You can also pass pushdown queries to read.jdbc

Answer 2

I use pissall's function (connect_to_sql) but I modified it a little.

from pyspark.sql import SparkSession

def connect_to_sql(
    spark, jdbc_hostname, jdbc_port, database, data_table, username, password
):
    jdbc_url = "jdbc:mysql://{0}:{1}/{2}".format(jdbc_hostname, jdbc_port, database)

    connection_details = {
        "user": username,
        "password": password,
        "driver": "com.mysql.jdbc.Driver",
    }

    df = spark.read.jdbc(url=jdbc_url, table=data_table, properties=connection_details)
    return df

if __name__=='__main__':
    spark = SparkSession \
        .builder \
        .appName('test') \
        .master('local[*]') \
        .enableHiveSupport() \
        .config("spark.driver.extraClassPath", <path to mysql-connector-java-5.1.49-bin.jar>) \
        .getOrCreate()

    df = connect_to_sql(spark, 'localhost', <port>, <database_name>, <table_name>, <user>, <password>)

or you can use SparkSession .read method

df = spark.read.format("jdbc").option("url","jdbc:mysql://localhost/<database_name>").option("driver","com.mysql.jdbc.Driver").option("dbtable",<table_name>).option("user",<user>).option("password",<password>).load()

Answer 3

import findspark
findspark.init("C:\ProgramData\Anaconda3\Lib\site-packages\pyspark")
import pyspark
import random
import os.path
from pyspark.sql.functions import *
import os
sc=pyspark.SparkContext(master="local[1]",appName="app1")
from pyspark.sql import SQLContext
sqlContext=SQLContext(sc)
spark = sqlContext.sparkSession.builder.appName("Spark reading jdbc").getOrCreate()
jdbcHostname = "srcmdepbd0\BI"
jdbcDatabase = "dbGestionFlota"
jdbcPort = 1433

jdbcUrl = "jdbc:sqlserver://{0}:{1};database={2}".format(jdbcHostname, jdbcPort, jdbcDatabase)
connectionProperties = {
  "integratedSecurity" : True,
  "driver" : "com.microsoft.sqlserver.jdbc.SQLServerDriver"
}
pushdown_query="SELECT TOP (1000) [iIDEstado],[iIDTipoEstado],[sEstado] ,[rOrden] ,[bAbierto],[sCampoAux1],[sCampoAux2]FROM [dbGestionFlota].[dbo].[tblSGSEstado]"
df1 = spark.read.jdbc(url=jdbcUrl, table=pushdown_query, properties=connectionProperties)
display(df1)

and this is my error

Py4JError                                 Traceback (most recent call last)
<ipython-input-15-a1c9096af940> in <module>
      1 pushdown_query="SELECT TOP (1000) [iIDEstado],[iIDTipoEstado],[sEstado] ,[rOrden] ,[bAbierto],[sCampoAux1],[sCampoAux2]FROM [dbGestionFlota].[dbo].[tblSGSEstado]"
----> 2 df1 = spark.read.jdbc(url=jdbcUrl, table=pushdown_query, properties=connectionProperties)
      3 display(df1)

c:\programdata\anaconda3\lib\site-packages\pyspark\sql\readwriter.py in jdbc(self, url, table, column, lowerBound, upperBound, numPartitions, predicates, properties)
    546         jprop = JavaClass("java.util.Properties", self._spark._sc._gateway._gateway_client)()
    547         for k in properties:
--> 548             jprop.setProperty(k, properties[k])
    549         if column is not None:
    550             assert lowerBound is not None, "lowerBound can not be None when ``column`` is specified"

C:\ProgramData\Anaconda3\Lib\site-packages\pyspark\python\lib\py4j-0.10.7-src.zip\py4j\java_gateway.py in __call__(self, *args)
   1255         answer = self.gateway_client.send_command(command)
   1256         return_value = get_return_value(
-> 1257             answer, self.gateway_client, self.target_id, self.name)
   1258 
   1259         for temp_arg in temp_args:

c:\programdata\anaconda3\lib\site-packages\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()

C:\ProgramData\Anaconda3\Lib\site-packages\pyspark\python\lib\py4j-0.10.7-src.zip\py4j\protocol.py in get_return_value(answer, gateway_client, target_id, name)
    330                 raise Py4JError(
    331                     "An error occurred while calling {0}{1}{2}. Trace:\n{3}\n".
--> 332                     format(target_id, ".", name, value))
    333         else:
    334             raise Py4JError(

Py4JError: An error occurred while calling o29.setProperty. Trace:
py4j.Py4JException: Method setProperty([class java.lang.String, class java.lang.Boolean]) does not exist
    at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:318)
    at py4j.reflection.ReflectionEngine.getMethod(ReflectionEngine.java:326)
    at py4j.Gateway.invoke(Gateway.java:274)
    at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
    at py4j.commands.CallCommand.execute(CallCommand.java:79)
    at py4j.GatewayConnection.run(GatewayConnection.java:238)
    at java.lang.Thread.run(Unknown Source)

any help?