Pyspark-Äquivalent von Pyodbc?

Question

Im Moment habe ich den folgenden Python-Code in einer Tabelle von einem lokalen SQL Server db in Pandas zu lesen:

import pandas as pd 
import pyodbc 

# Connect to DB 
server = 'server' 
db = 'db' 

conn = pyodbc.connect('DRIVER={SQL SERVER}; SERVER=' + server + '; DATABASE=' + db + '; TRUSTED_CONNECTION=yes') 
cursor = conn.cursor() 

table = 'table' 
df = pd.read_sql('Select * From ' + table, conn) 

Dieser Code funktioniert, aber jetzt würde Ich mag die gleiche Sache in Pyspark zu tun. Was ist dieser Code in Pyspark?

Ich habe versucht, die folgenden:

import findspark 
import os 
from pyspark.sql import SparkSession 
from pyspark.sql.functions import * 

# didn't know which of these would work so tried both 
os.environ['SPARK_CLASSPATH'] = 'path/to/sqljdbc42.jar' 
os.environ['driver-class-path'] = 'path/to/sqljdbc42.jar' 

findspark.init('C:/spark/spark') 

spark = SparkSession \ 
     .builder \ 
     .appName("SparkCoreTest") \ 
     .getOrCreate() 

sc = spark.sparkContext 
sqlctx = SQLContext(sc) 

server = 'server' 
db = 'db' 
url = 'jdbc:sqlserver//' + server + ';databaseName=' + db 
table = 'table' 
properties = {'driver' : 'com.microsoft.sqlserver.jdbc.SQLServerDriver'} 

df = sqlctx.read.format('jdbc').options(url=url, dbtable=table, driver='{SQL SERVER}').load() 

Dies gibt java.lang.ClassNotFoundException: {SQL SERVER}. Während dieses Prozesses habe ich auch einen Fehler bekommen, weil ich keinen "passenden Treiber" finden konnte, obwohl ich denke, dass ich das durch Ändern von os.environ behoben habe. Jede Hilfe würde sehr geschätzt werden!

Answer 1

Sie sind ziemlich nah. Das ClassNotFoundException bedeutet, dass es das Treiberglas nicht finden kann.

Um Treiber für MySQL, SQL Server usw. zu verwenden, ist es wichtig, die JARs in einem Ordner zu haben, auf den sowohl Master als auch Slaves zugreifen können (wenn Sie Spark lokal ausführen, dann sind Sie in Ordnung) und Sie müssen angeben die Lage der Gläser entweder mit der --jars Flagge wie folgt aus:

spark-submit --jars /path/to/sqljdbc42.jar ...other params... yourscript.py

oder:

# create a configuration instance and set the spark.jars option 
conf = SparkConf() 
conf.set('spark.jars', '/path/to/sqljdbc42.jar') 

# give that configuration to the spark session 
spark = SparkSession.builder\ 
        .config(conf=conf) \ 
        .appName(app_name) \ 
        .getOrCreate() 

Darüber hinaus sollten Sie die Funken Sitzung nicht die SqlContext direkt von den s nutzen zu lesen ql-Server (auch gedacht, die auf dem Funken Version abhängig):

df = spark.read.format('jdbc').options(url=url, dbtable=table).load() 

(driver jar )

(Anmerkung: Es gibt auch eine Möglichkeit, mit Umgebungsvariablen zu arbeiten, aber ich habe es nicht verwendet wird)

Hoffe, das hilft, viel Glück!