現在我們來實現在自定義程式中編寫Spark SQL查詢程式。

實現查詢的方式有兩種：

方式一：通過反射推斷schema。

方式二：通過structtype直接指定schema。

這次我們用方式二來實現自定義查詢。

具體程式如下：

package cn.allengao.sparksql

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

object SpecifyingSchema {
  def main(args: Array[String]): Unit = {
    //1、模板程式碼
    val conf = new SparkConf().setAppName("SQL-typeTwo").setMaster("local[*]")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    //2、獲取資料
    val linesRDD = sc.textFile(args(0)).map(_.split(","))
    //3、通過StructType指定每個欄位的Schema
    val schema = StructType(
      List(
        StructField("id",IntegerType,true),
        StructField("name",StringType,true),
        StructField("colligate",IntegerType,true),
        StructField("attack",IntegerType,true),
        StructField("defense",IntegerType,true)
      )
    )
    //4、將RDD對映到rowRDD
    val rowRDD = linesRDD.map(x=>Row(x(0).toInt,x(1),x(2).toInt,x(3).toInt,x(4).toInt))
    val teamDF = sqlContext.createDataFrame(rowRDD,schema)
    //5、登錄檔
    teamDF.registerTempTable("t_team")
    //6、查詢
    val df = sqlContext.sql("select * from t_team order by colligate desc limit 3")
    //7、輸出
    df.write.json(args(1))

    sc.stop()
  }
}
 

對比總結：

1.case class模板類模式視覺化比較好。

2.case class模板類引數上限為22個，對於欄位多的不能使用。

3.程式設計方式更適合日常開發。