Apache SparkSQL 用JDBC方式查询MySQL数据库示例_using org.apache.spark.sql.jdbc

第一步，初始化SparkSession

语法：

val sparkSession = SparkSession.builder()
      .master("local[*]")
      .appName("Spark SQL JDBC")
      .getOrCreate()

1
2
3
4
5

第二步，JDBC读取mysql数据为DataFrame

语法：

val jdbcDF = sparkSession.read
      .format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306/bigdata0408?characterEncoding=UTF-8")
      .option("dbtable", "paydata")
      .option("user", "root")
      .option("password", "root")
      .load()
1
2
3
4
5
6
7

第三步，根据DataFrame创建一个TempView

TempView意为临时的视图，view，它可以使用sql方法进行SQL操作。
语法：

jdbcDF.createOrReplaceTempView("paydata")
1

完整代码示例

package demo11

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Row, SparkSession}

object Test02 {
  def main(args: Array[String]): Unit = {
    val sparkSession = SparkSession.builder()
      .master("local[*]")
      .appName("Spark SQL JDBC")
      .getOrCreate()

    val jdbcDF = sparkSession.read
      .format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306/bigdata0408?characterEncoding=UTF-8")
      .option("dbtable", "paydata")
      .option("user", "root")
      .option("password", "root")
      .load()

    jdbcDF.createOrReplaceTempView("paydata")

    //18、计算出商户名称为33333328的所有金额为多少？
    sparkSession.sql("SELECT sum(money) FROM paydata WHERE posname == '33333328'").show()


    //19、计算出支付通道为alipay的所有金额？
    sparkSession.sql("SELECT sum(money) FROM paydata WHERE paychannel == 'alipay'").show()

  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32