十一、spark SQL的scala示例

簡介

spark SQL官網：http://spark.apache.org/docs/latest/sql-programming-guide.html

sparkSQL是構建在sparkCore之上的組件，用於處理結構化的數據。它將數據抽象為DataFrame並提供豐富的API，並且sparkSQL允許使用SQL腳本進行操作，使得數據查詢變得非常的容易使用。

同時，sparkSQL除了操作簡單，API豐富之外，對於數據源的支持也很強大。你可以從，如：

1）HDFS

2）Parguet文件

3）json文件

4）JDBC

5）ODBC

6）HIVE

等多種數據源來創建dataFrame，也可以從spark的RDD轉換成dataFrame。

代碼示例

下面是scala的代碼示例：

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SQLContext}
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
import org.apache.spark.{SparkConf, SparkContext}


/**
  * @Description spark sql demo
  * @Author lay
  * @Date 2018/12/09 21:33
  */
object SparkSQLDemo {
  var conf: SparkConf = _
  var sc: SparkContext = _
  var userData: Array[String] = Array("1 lay 23", "2 marry 24", "3 gary 25")
  var userRDD: RDD[Row] = _
  var sqlContext: SQLContext = _
  var df: DataFrame = _

  def init(): Unit = {
    conf = new SparkConf().setAppName("spark sql demo").setMaster("local")
    sc = new SparkContext(conf)
    // 創建sqlContext
    sqlContext = new SQLContext(sc)
    // 創建schema
    var structFields = Array(StructField("id", IntegerType), StructField("name", StringType), StructField("age", IntegerType))
    var schema = new StructType(structFields)
    // 創建RDD
    userRDD = sc.parallelize(userData).map{x => val lines = x.split(" ");Row(lines(0).toInt, lines(1), lines(2).toInt)}
    // 創建dataFrame
    df = sqlContext.createDataFrame(userRDD, schema)
  }

  def main(args: Array[String]): Unit = {
    init()
    // dataFrame方式查詢：查詢年齡大於23歲的用戶的姓名
    df.select("name").where("age > 23").show()
    // 注冊為t_user表
    df.createOrReplaceTempView("t_user")
    // SQL方式查詢：年齡大於23歲的用戶的姓名
    sqlContext.sql("SELECT name FROM t_user WHERE age > 23").show()
  }
}

以上代碼將RDD通過StructType轉換成了dataFrame，然后分別采用dataFrame的API和SQL兩種方式查詢出了結果，如圖：