从星火连接到mysql

Question

我正试着按照这里提到的指示..。

https://www.percona.com/blog/2016/08/17/apache-spark-makes-slow-mysql-queries-10x-faster/

而这里..。

https://www.percona.com/blog/2015/10/07/using-apache-spark-mysql-data-analysis/

我用的是火花码头的图像。

docker run -it -p 8088:8088 -p 8042:8042 -p 4040:4040 -h sandbox sequenceiq/spark:1.6.0 bash

cd /usr/local/spark/

./sbin/start-master.sh

./bin/spark-shell --driver-memory 1G --executor-memory 1g --executor-cores 1 --master local

这项工作如预期的那样：

scala> sc.parallelize(1 to 1000).count()

但这表明了一个错误：

val jdbcDF = spark.read.format("jdbc").options(
  Map("url" ->  "jdbc:mysql://1.2.3.4:3306/test?user=dba&password=dba123",
  "dbtable" -> "ontime.ontime_part",
  "fetchSize" -> "10000",
  "partitionColumn" -> "yeard", "lowerBound" -> "1988", "upperBound" -> "2016", "numPartitions" -> "28"
  )).load()

这是一个错误：

<console>:25: error: not found: value spark
         val jdbcDF = spark.read.format("jdbc").options(

如何从火花外壳内连接到MySQL？

Answer 1

使用Spark2.0.x，您可以使用DataFrameReader和DataFrameWriter。使用SparkSession.read访问DataFrameReader，使用Dataset.write访问DataFrameWriter。

假设使用火花壳。

读取示例

val prop=new java.util.Properties()
prop.put("user","username")
prop.put("password","yourpassword")
val url="jdbc:mysql://host:port/db_name"

val df=spark.read.jdbc(url,"table_name",prop) 
df.show()

阅读示例2

val jdbcDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:mysql:dbserver")
  .option("dbtable", “schema.tablename")
  .option("user", "username")
  .option("password", "password")
  .load()

来自火花医生

写例

import org.apache.spark.sql.SaveMode

val prop=new java.util.Properties()
prop.put("user","username")
prop.put("password","yourpassword")
val url="jdbc:mysql://host:port/db_name"
//df is a dataframe contains the data which you want to write.
df.write.mode(SaveMode.Append).jdbc(url,"table_name",prop)

Answer 2

看起来spark没有定义，您应该使用SQLContext连接到驱动程序，如下所示：

import org.apache.spark.sql.SQLContext

val sqlcontext = new org.apache.spark.sql.SQLContext(sc)

val dataframe_mysql = sqlcontext.read.format("jdbc").option("url", "jdbc:mysql://Public_IP:3306/DB_NAME").option("driver", "com.mysql.jdbc.Driver").option("dbtable", "tblage").option("user", "sqluser").option("password", "sqluser").load()

稍后，您可以在使用spark的地方使用sqlcontext (在spark.read等中)

Answer 3

• 首先创建火花上下文
• 确保将jdbc jar文件附加到类路径中。

如果您试图从jdbc读取数据。使用dataframe代替RDD，因为dataframes具有更好的性能。请参阅下面的性能比较图。

​

​

下面是从jdbc读取数据的语法

SparkConf conf = new SparkConf().setAppName("app"))
                .setMaster("local[2]")
                .set("spark.serializer",prop.getProperty("spark.serializer"));

JavaSparkContext sc = new JavaSparkContext(conf);
sqlCtx = new SQLContext(sc);

df = sqlCtx.read()
                        .format("jdbc")
                        .option("url", "jdbc:mysql://1.2.3.4:3306/test")
                        .option("driver", "com.mysql.jdbc.Driver")
                        .option("dbtable","dbtable")
                        .option("user", "dbuser")
                        .option("password","dbpwd"))
                        .load();