Pyspark 连接到 Microsoft SQL 服务器?
Posted
技术标签:
【中文标题】Pyspark 连接到 Microsoft SQL 服务器?【英文标题】:Pyspark connection to the Microsoft SQL server? 【发布时间】:2019-11-14 00:25:10 【问题描述】:我在 SQL server 中有一个庞大的数据集,我想用 python 连接 SQL server,然后使用 pyspark 运行查询。
我已经看过 JDBC 驱动程序,但我没有找到方法,我使用 PYODBC 但没有使用 spark。
任何帮助将不胜感激。
【问题讨论】:
【参考方案1】:请使用以下方式连接 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
是一个SparkSession
对象,其余的都很清楚。
您也可以将下推查询传递给read.jdbc
【讨论】:
感谢您的回答,我想我已经接近解决它了,但是我遇到了 spark session 的问题。 感谢您的回答,我的 sparksession 出现错误,我所拥有的是:spark = SparkSession.builder\ .config('spark.jars', r'D:\Usuarios\ aospinam\Downloads\Microsoft JDBC Driver 7.4 for SQL Server\sqljdbc_7.4\enu\mssql-jdbc-7.4.1.jre8')\.appName("app2")\.master("local").getOrCreate() @AndresOspina 你不需要添加罐子。您可以摆脱该配置 我如何知道我的 SparkSession 是否建立良好? @AndresOspinaSparkSession
getOrCreate
不是一个惰性求值函数。所以当你创建它时,你应该立即得到错误。让我知道错误是什么。【参考方案2】:
我使用了 pissall 的函数(connect_to_sql),但我稍微修改了一下。
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>)
或者你可以使用 SparkSession .read 方法
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()
【讨论】:
【参考方案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)
这是我的错误
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 012. Trace:\n3\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)
有什么帮助吗?
【讨论】:
这不是答案。如果您有更多详细信息要添加,请编辑您的原始问题。以上是关于Pyspark 连接到 Microsoft SQL 服务器?的主要内容,如果未能解决你的问题,请参考以下文章
Pyspark/SQL 将具有列表值的列连接到另一个数据框列
连接到 sql server 时出现 pyspark jdbc 错误
无法使用 Jupyter 笔记本上的 pyspark 从 Apache Spark 连接到 MS SQL
Pyspark:使用 Python 从 Spark 2.4 连接到 MS SQL Server 2017 时没有合适的驱动程序错误