Spark學(xué)習(xí)筆記之Spark SQL的具體使用

更新時(shí)間：2019年06月14日 10:12:28 作者：EVAO_大個(gè)子

這篇文章主要介紹了Spark學(xué)習(xí)筆記之Spark SQL的具體使用,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧

1. Spark SQL是什么?

處理結(jié)構(gòu)化數(shù)據(jù)的一個(gè)spark的模塊
它提供了一個(gè)編程抽象叫做DataFrame并且作為分布式SQL查詢(xún)引擎的作用

2. Spark SQL的特點(diǎn)

多語(yǔ)言的接口支持(java python scala)
統(tǒng)一的數(shù)據(jù)訪問(wèn)
完全兼容hive
支持標(biāo)準(zhǔn)的連接

3. 為什么學(xué)習(xí)SparkSQL?

我們已經(jīng)學(xué)習(xí)了Hive，它是將Hive SQL轉(zhuǎn)換成MapReduce然后提交到集群上執(zhí)行，大大簡(jiǎn)化了編寫(xiě)MapReduce的程序的復(fù)雜性，由于MapReduce這種計(jì)算模型執(zhí)行效率比較慢。所有Spark SQL的應(yīng)運(yùn)而生，它是將Spark SQL轉(zhuǎn)換成RDD，然后提交到集群執(zhí)行，執(zhí)行效率非?？欤?/p>

4. DataFrame(數(shù)據(jù)框)

與RDD類(lèi)似，DataFrame也是一個(gè)分布式數(shù)據(jù)容器
然而DataFrame更像傳統(tǒng)數(shù)據(jù)庫(kù)的二維表格，除了數(shù)據(jù)以外，還記錄數(shù)據(jù)的結(jié)構(gòu)信息，即schema
DataFrame其實(shí)就是帶有schema信息的RDD

5. SparkSQL1.x的API編程

<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-sql_2.11</artifactId>
  <version>${spark.version}</version>
</dependency>

5.1 使用sqlContext創(chuàng)建DataFrame(測(cè)試用)

object Ops3 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("Ops3").setMaster("local[3]")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)
    val rdd1 = sc.parallelize(List(Person("admin1", 14, "man"),Person("admin2", 16, "man"),Person("admin3", 18, "man")))
    val df1: DataFrame = sqlContext.createDataFrame(rdd1)
    df1.show(1)
  }
}
case class Person(name: String, age: Int, sex: String);

5.2 使用sqlContxet中提供的隱式轉(zhuǎn)換函數(shù)(測(cè)試用)

import org.apache.spark
val conf = new SparkConf().setAppName("Ops3").setMaster("local[3]")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val rdd1 = sc.parallelize(List(Person("admin1", 14, "man"), Person("admin2", 16, "man"), Person("admin3", 18, "man")))
import sqlContext.implicits._
val df1: DataFrame = rdd1.toDF
df1.show()
5.3 使用SqlContext創(chuàng)建DataFrame(常用)
val conf = new SparkConf().setAppName("Ops3").setMaster("local[3]")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val linesRDD: RDD[String] = sc.textFile("hdfs://uplooking02:8020/sparktest/")
val schema = StructType(List(StructField("name", StringType), StructField("age", IntegerType), StructField("sex", StringType)))
val rowRDD: RDD[Row] = linesRDD.map(line => {
 val lineSplit: Array[String] = line.split(",")
 Row(lineSplit(0), lineSplit(1).toInt, lineSplit(2))
})
val rowDF: DataFrame = sqlContext.createDataFrame(rowRDD, schema)
rowDF.show()

6. 使用新版本的2.x的API

val conf = new SparkConf().setAppName("Ops5") setMaster ("local[3]")
val sparkSession: SparkSession = SparkSession.builder().config(conf).getOrCreate()
val sc = sparkSession.sparkContext
val linesRDD: RDD[String] = sc.textFile("hdfs://uplooking02:8020/sparktest/")
//數(shù)據(jù)清洗
val rowRDD: RDD[Row] = linesRDD.map(line => {
  val splits: Array[String] = line.split(",")
  Row(splits(0), splits(1).toInt, splits(2))
})
val schema = StructType(List(StructField("name", StringType), StructField("age", IntegerType), StructField("sex", StringType)))
val df: DataFrame = sparkSession.createDataFrame(rowRDD, schema)

df.createOrReplaceTempView("p1")
val df2 = sparkSession.sql("select * from p1")
df2.show()

7. 操作SparkSQL的方式

7.1 使用SQL語(yǔ)句的方式對(duì)DataFrame進(jìn)行操作

val conf = new SparkConf().setAppName("Ops5") setMaster ("local[3]")
val sparkSession: SparkSession = SparkSession.builder().config(conf).getOrCreate()//Spark2.x新的API相當(dāng)于Spark1.x的SQLContext
val sc = sparkSession.sparkContext
val linesRDD: RDD[String] = sc.textFile("hdfs://uplooking02:8020/sparktest/")
//數(shù)據(jù)清洗
val rowRDD: RDD[Row] = linesRDD.map(line => {
  val splits: Array[String] = line.split(",")
  Row(splits(0), splits(1).toInt, splits(2))
})
val schema = StructType(List(StructField("name", StringType), StructField("age", IntegerType), StructField("sex", StringType)))
val df: DataFrame = sparkSession.createDataFrame(rowRDD, schema)

df.createOrReplaceTempView("p1")//這是Sprk2.x新的API 相當(dāng)于Spark1.x的registTempTable()
val df2 = sparkSession.sql("select * from p1")
df2.show()

7.2 使用DSL語(yǔ)句的方式對(duì)DataFrame進(jìn)行操作

DSL(domain specific language ) 特定領(lǐng)域語(yǔ)言

val conf = new SparkConf().setAppName("Ops5") setMaster ("local[3]")
val sparkSession: SparkSession = SparkSession.builder().config(conf).getOrCreate()
val sc = sparkSession.sparkContext
val linesRDD: RDD[String] = sc.textFile("hdfs://uplooking02:8020/sparktest/")
//數(shù)據(jù)清洗
val rowRDD: RDD[Row] = linesRDD.map(line => {
  val splits: Array[String] = line.split(",")
  Row(splits(0), splits(1).toInt, splits(2))
})
val schema = StructType(List(StructField("name", StringType), StructField("age", IntegerType), StructField("sex", StringType)))
val rowDF: DataFrame = sparkSession.createDataFrame(rowRDD, schema)
import sparkSession.implicits._
val df: DataFrame = rowDF.select("name", "age").where("age>10").orderBy($"age".desc)
df.show()

8. SparkSQL的輸出

8.1 寫(xiě)出到JSON文件

val conf = new SparkConf().setAppName("Ops5") setMaster ("local[3]")
val sparkSession: SparkSession = SparkSession.builder().config(conf).getOrCreate()
val sc = sparkSession.sparkContext
val linesRDD: RDD[String] = sc.textFile("hdfs://uplooking02:8020/sparktest")
//數(shù)據(jù)清洗
val rowRDD: RDD[Row] = linesRDD.map(line => {
  val splits: Array[String] = line.split(",")
  Row(splits(0), splits(1).toInt, splits(2))
})
val schema = StructType(List(StructField("name", StringType), StructField("age", IntegerType), StructField("sex", StringType)))
val rowDF: DataFrame = sparkSession.createDataFrame(rowRDD, schema)
import sparkSession.implicits._
val df: DataFrame = rowDF.select("name", "age").where("age>10").orderBy($"age".desc)
df.write.json("hdfs://uplooking02:8020/sparktest1")

8.2 寫(xiě)出到關(guān)系型數(shù)據(jù)庫(kù)(mysql)

val conf = new SparkConf().setAppName("Ops5") setMaster ("local[3]")
val sparkSession: SparkSession = SparkSession.builder().config(conf).getOrCreate()
val sc = sparkSession.sparkContext
val linesRDD: RDD[String] = sc.textFile("hdfs://uplooking02:8020/sparktest")
//數(shù)據(jù)清洗
val rowRDD: RDD[Row] = linesRDD.map(line => {
  val splits: Array[String] = line.split(",")
  Row(splits(0), splits(1).toInt, splits(2))
})
val schema = StructType(List(StructField("name", StringType), StructField("age", IntegerType), StructField("sex", StringType)))
val rowDF: DataFrame = sparkSession.createDataFrame(rowRDD, schema)
import sparkSession.implicits._
val df: DataFrame = rowDF.select("name", "age").where("age>10").orderBy($"age".desc)
val url = "jdbc:mysql://localhost:3306/test"
//表會(huì)自動(dòng)創(chuàng)建
val tbName = "person1";
val prop = new Properties()
prop.put("user", "root")
prop.put("password", "root")
//SaveMode 默認(rèn)為ErrorIfExists
df.write.mode(SaveMode.Append).jdbc(url, tbName, prop)

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: