python - PySpark 到 MySQL 插入错误？

Question

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

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

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

它说“java.lang.RuntimeException:[1.5]失败:``.''预期但发现':'”

此时:

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

数据库名称为“bigdata”

上面包含用户名和密码

我认为端口号是正确的

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

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 提交声明:

./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 时发生错误，未找到合适的驱动程序”。

关于这个有什么想法吗？

Answer 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')