如何在 Jupyter notebook 中为 pyspark 设置 MySQL 的 JDBC 驱动程序？

Posted 2023-03-27

【中文标题】如何在 Jupyter notebook 中为 pyspark 设置 MySQL 的 JDBC 驱动程序？

【英文标题】：How to set up JDBC driver for MySQL in Jupyter notebook for pyspark?

【发布时间】：2020-08-31 19:35:54

【问题描述】：

我正在尝试将一堆 CSV 文件逐行加载到使用 pyspark 配置在 OpenShift 上运行的 mysql 实例中。我有一个启动并运行 Spark 的 Jupyter 笔记本。

下面是我的代码。并且由于特定的驱动程序错误而失败

Py4JJavaError: An error occurred while calling o89.save.

from pyspark.sql import SparkSession
from pyspark.sql import SQLContext

if __name__ == '__main__':
    scSpark = SparkSession \
        .builder \
        .appName("reading csv") \
        .getOrCreate()

if __name__ == '__main__':
    scSpark = SparkSession \
        .builder \
        .appName("reading csv") \
        .getOrCreate()
    
data_file = '/opt/app-root/src/data/train.psv'
sdfData = scSpark.read.csv(data_file, header=True, sep="|").cache()
print('Total Records = '.format(sdfData.count()))
sdfData.show()

sdfData.registerTempTable("train")
output =  scSpark.sql('SELECT count(*) from train')
output.show()

+--------+
|count(1)|
+--------+
| 1168686|
+--------+

import os
os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages mysql:mysql-connector-java:jar:8.0.21 pyspark-shell'

output = scSpark.sql('SELECT * from train')
output.show()
output.write.format('jdbc').options(
    url='jdbc:mysql://mysql-1-28d85/sepsis',
    driver='com.mysql.jdbc.Driver',
    #driver='mysql-connector-java.Driver',
    #driver='org.mysql.jdbc.Driver',
    dbtable='train',
    user='sepsis',
    password='Success_2020').mode('append').save()


---------------------------------------------------------------------------
Py4JJavaError                             Traceback (most recent call last)
<ipython-input-57-114af97e0442> in <module>
     11     dbtable='train',
     12     user='sepsis',
---> 13     password='Success_2020').mode('append').save()

/opt/app-root/lib/python3.6/site-packages/pyspark/sql/readwriter.py in save(self, path, format, mode, partitionBy, **options)
    735             self.format(format)
    736         if path is None:
--> 737             self._jwrite.save()
    738         else:
    739             self._jwrite.save(path)

/opt/app-root/lib/python3.6/site-packages/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:

/opt/app-root/lib/python3.6/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()

/opt/app-root/lib/python3.6/site-packages/py4j/protocol.py in get_return_value(answer, gateway_client, target_id, name)
    326                 raise Py4JJavaError(
    327                     "An error occurred while calling 012.\n".
--> 328                     format(target_id, ".", name), value)
    329             else:
    330                 raise Py4JError(

Py4JJavaError: An error occurred while calling o1641.save.
: java.lang.ClassNotFoundException: com.mysql.jdbc.Driver
    at java.net.URLClassLoader.findClass(URLClassLoader.java:382)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:419)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:352)
    at org.apache.spark.sql.execution.datasources.jdbc.DriverRegistry$.register(DriverRegistry.scala:45)
    at org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions$$anonfun$5.apply(JDBCOptions.scala:99)
    at org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions$$anonfun$5.apply(JDBCOptions.scala:99)
    at scala.Option.foreach(Option.scala:257)
    at org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions.<init>(JDBCOptions.scala:99)
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcOptionsInWrite.<init>(JDBCOptions.scala:190)
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcOptionsInWrite.<init>(JDBCOptions.scala:194)
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcRelationProvider.createRelation(JdbcRelationProvider.scala:45)
    at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala:45)
    at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:70)
    at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:68)
    at org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(commands.scala:86)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:131)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:127)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:155)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
    at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
    at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:127)
    at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:83)
    at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:81)
    at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:676)
    at org.apache.spark.sql.DataFrameWriter$$anonfun$runCommand$1.apply(DataFrameWriter.scala:676)
    at org.apache.spark.sql.execution.SQLExecution$$anonfun$withNewExecutionId$1.apply(SQLExecution.scala:80)
    at org.apache.spark.sql.execution.SQLExecution$.withSQLConfPropagated(SQLExecution.scala:127)
    at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:75)
    at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala:676)
    at org.apache.spark.sql.DataFrameWriter.saveToV1Source(DataFrameWriter.scala:285)
    at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:271)
    at sun.reflect.GeneratedMethodAccessor67.invoke(Unknown Source)
    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:282)
    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(Thread.java:748)

用包更改了代码。

这也是 openshift ，所有组件都作为 pod 运行，无法访问外部环境。

【参考方案1】：

java.lang.ClassNotFoundException: com.mysql.cj.jdbc.Driver

这说明了一切。您必须使用--driver-class-path 或类似（特定于 Jupyter）的 MySQL JDBC 驱动程序启动 pyspark（或环境）。

对于 Jupyter 笔记本

从PySpark in Jupyter Notebook — Working with Dataframe & JDBC Data Sources复制：

如果你使用 Jupyter Notebook，你应该设置PYSPARK_SUBMIT_ARGS 环境变量，如下：

import os
os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages org.postgresql:postgresql:42.1.1 pyspark-shell'

更改 --packages 以引用 MySQL JDBC 驱动程序。

【讨论】：

我已经尝试了这些方法中的每一种，但它总是以同样的错误失败。 import os os.environ['PYSPARK_SUBMIT_ARGS'] = '--packages mysql:mysql-connector- java:jar:8.0.21 pyspark-shell'

@NIkhilMehta 你能编辑你的问题并用import和--packages添加单元格的输出吗？查看 Jupyter 日志以了解更多信息。由于--packages 尝试从公共存储库访问 jar，一个明显的问题是“您是否可以从 env 访问 Internet？”

使用 --packages 的输出编辑的代码。由于这都是 Open Shift 和安全环境，因此 Pod 内无法访问互联网。

【参考方案2】：

进入spark的安装路径后，会有一个jars文件夹。下载你的 mysql jdbc jar 文件并将其放入 jars 文件夹，然后你不需要任何命令或代码选项。

【讨论】：

作为一个 Open Shift 环境，所有这些容器都作为不可变的 Pod 运行。所以不能将罐子复制到环境中。