如何将 Spark SQL 与 My SQL 数据库 Scala 连接起来

Question

问题陈述：

嗨，我是 Spark World 的新手。我想查询 MySQL 数据库，然后将一张表加载到 Spark 中。然后我想使用 SQL Query 在表上应用一些过滤器。过滤结果后，我想将结果作为 JSON 返回。所有这些我们都必须在一个独立的 Scala 基础应用程序中完成。

我正在努力初始化 Spark 上下文并收到错误消息。我知道我遗漏了一些信息。

有人可以看看代码并告诉我我需要做什么。

代码：

import application.ApplicationConstants
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{SparkSession, Dataset, Row, Column, SQLContext}

var sc: SparkContext = null

    val sparkSession = SparkSession.builder().master("spark://10.62.10.71:7077")
      .config("format","jdbc")
      .config("url","jdbc:mysql://localhost:3306/test")
      .config("user","root")
      .config("password","")
      .appName("MySQLSparkConnector")
      .getOrCreate()

    var conf = new SparkConf()
    conf.setAppName("MongoSparkConnectorIntro")
      .setMaster("local")
      .set("format", "jdbc")
      .set("url","jdbc:mysql://localhost:3306/test")
      .set("user","root")
      .set("password","")

    sc = new SparkContext(conf)
val connectionProperties = new java.util.Properties
    connectionProperties.put("user", username)
    connectionProperties.put("password", password)
     val customDF2 = sparkSession.read.jdbc(url,"employee",connectionProperties)

    println("program ended")

错误：

以下是我得到的错误：

64564 [main] ERROR org.apache.spark.SparkContext - Error initializing SparkContext.
java.lang.NullPointerException
    at org.apache.spark.SparkContext.<init>(SparkContext.scala:560)
    at org.apache.spark.SparkContext$.getOrCreate(SparkContext.scala:2520)
    at org.apache.spark.sql.SparkSession$Builder.$anonfun$getOrCreate$5(SparkSession.scala:935)
    at scala.Option.getOrElse(Option.scala:121)
    at org.apache.spark.sql.SparkSession$Builder.getOrCreate(SparkSession.scala:926)
    at manager.SparkSQLMySQLDBConnector$.main(SparkSQLMySQLDBConnector.scala:21)
    at manager.SparkSQLMySQLDBConnector.main(SparkSQLMySQLDBConnector.scala)
64566 [main] INFO org.apache.spark.SparkContext - SparkContext already stopped.
Exception in thread "main" java.lang.NullPointerException
    at org.apache.spark.SparkContext.<init>(SparkContext.scala:560)
    at org.apache.spark.SparkContext$.getOrCreate(SparkContext.scala:2520)
    at org.apache.spark.sql.SparkSession$Builder.$anonfun$getOrCreate$5(SparkSession.scala:935)
    at scala.Option.getOrElse(Option.scala:121)
    at org.apache.spark.sql.SparkSession$Builder.getOrCreate(SparkSession.scala:926)
    at manager.SparkSQLMySQLDBConnector$.main(SparkSQLMySQLDBConnector.scala:21)
    at manager.SparkSQLMySQLDBConnector.main(SparkSQLMySQLDBConnector.scala)

P.S：如果有人可以给我任何显示 Scala 类似场景的链接或教程。

版本：

火花：2.4.0 斯卡拉：2.12.8 MySQL 连接器 Jar：8.0.13

Answer 1

我认为你在搞乱创建 spark 上下文和配置来连接 MYSQL

如果您使用的是 spark 2.0+，则仅使用 SparkSession 作为入口点

val spark = SparkSession.builder().master("local[*]").appName("Test").getOrCreate

//Add Properties asbelow  
val prop = new java.util.Properties()
prop.put("user", "user")
prop.put("password", "password")
val url = "jdbc:mysql://host:port/dbName"

现在用数据框读取表格

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

要访问sparkContext 和sqlContext，您可以从SparkSession 访问

val sc = spark.sparkContext

val sqlContext = spark.sqlContext

确保您在类路径中有mysql-connector-java jar，将依赖项添加到您的pom.xml 或built.sbt

希望这会有所帮助！