PySpark 到 MySQL 插入错误？答案

【问题标题】：PySpark to MySQL Insert Error?PySpark 到 MySQL 插入错误？
【发布时间】：2017-05-02 18:11:46
【问题描述】：

我正在尝试学习 PySpark 并编写了一个简单的脚本，该脚本从我的一个 HDFS 目录中加载一些 JSON 文件，将每个文件加载为 python 字典（使用 json.loads() ），然后为每个对象提取一些字段。

相关信息存储在 Spark Dataframe 中，我想将此数据插入 MySQL 表（我在本地创建）。

但是，当我运行它时，我的连接 URL 出现错误。

它说“java.lang.RuntimeException: [1.5] failure: `.'' 预期但 `:' 找到”

此时：

jdbc:mysql://localhost:3306/bigdata?user=root&password=pwd
    ^

数据库名称是“大数据”

上面有用户名和密码

我认为正确的端口号

这是我的完整脚本......：

import json 
import pandas as pd 
import numpy as np 
from pyspark import SparkContext 
from pyspark.sql import Row, SQLContext

SQL_CONNECTION="jdbc:mysql://localhost:3306/bigdata?user=root&password=pwd"
sc=SparkContext()
sqlContext = SQLContext(sc)

cols = ['Title', 'Site']
df = pd.DataFrame(columns=cols)

#First, load my files as RDD and convert them as JSON
rdd1 = sc.wholeTextFiles("hdfs://localhost:8020/user/ashishu/Project/sample data/*.json")
rdd2 = rdd1.map(lambda kv: json.loads(kv[1])) 

#Read in the RDDs and do stuff
for record in rdd2.take(2):

    title = record['title']
    site = record['thread']['site_full']
    vals = np.array([title, site])
    df.loc[len(df)] = vals

sdf = sqlContext.createDataFrame(df)
sdf.show()
sdf.insertInto(SQL_CONNECTION, "sampledata")

SQL_CONNECTION 是开头的连接 URL，“sampledata”是我要在 MySQL 中插入的表的名称。要使用的特定数据库在连接 url（“bigdata”）中指定。

这是我的 spark-submit 声明：

./bin/spark-submit /Users/ashishu/Desktop/sample.py --driver-class-path /Users/ashishu/Documents/Spark/.../bin/mysql-connector-java-5.1.42/mysql-connector-java-5.1.42-bin.jar

我使用的是 Spark 1.6.1

我在这里错过了关于 MySQL 连接的一些愚蠢的东西吗？我尝试用“。”替换“：”（在jdbc和mysql之间）。但这显然没有解决任何问题并产生了不同的错误...

谢谢

编辑

我根据建议修改了我的代码，这样我就没有使用 sdf.InsertInto，而是说...

sdf.write.jdbc(SQL_CONNECTION, table="sampledata", mode="append")

但是，现在我在终端中使用以下提交命令后出现新错误：

./bin/spark-submit sample.py --jars <path to mysql-connector-java-5.1.42-bin.jar>

错误基本上是说“调用o53.jdbc时发生错误，找不到合适的驱动程序”。

关于这个有什么想法吗？

【问题讨论】：

标签： python mysql hadoop apache-spark pyspark

【解决方案1】：

insertInto 期望 tablename 或 database.tablename 这就是它抛出 . expected but : found 错误的原因。您需要的是jdbc 数据帧编写器，即参见此处http://spark.apache.org/docs/latest/api/python/pyspark.sql.html#pyspark.sql.DataFrameWriter.jdbc

类似-

sdf.write.jdbc(SQL_CONNECTION, table=bigdata.sampledata,mode='append')

【讨论】：

谢谢，所以当我尝试此操作时（请参阅编辑以发布我的全部更改），我收到一个新错误，提示“调用 o53.jdbc 时发生错误没有合适的驱动程序”。你以前遇到过吗？
这可能会有所帮助 - stackoverflow.com/questions/36588084/…
我也尝试过使用 --packages mysql:mysql-connector-java:5.1.42 但我遇到了同样的问题。如何配置 mysql-connector 以便将其识别为包？也许这就是问题所在。谢谢。

【解决方案2】：

我想通了，解决方案是在我的 /spark/conf 文件夹中创建一个 spark-env.sh 文件，并在其中进行以下设置：

export SPARK_CLASSPATH=$SPARK_CLASSPATH:/<path to your mysql connector jar file>

谢谢！

【讨论】：