?1、創(chuàng)建Spark Session

val spark = SparkSession.builder. master("local").appName("spark session example").getOrCreate()
注：下面的 spark 都指的是 sparkSession

2、將RDD隱式轉(zhuǎn)換為DataFrame

import spark.implicits._ 3、SparkSession 介紹 spark中包含 sparkContext和 sqlContext兩個(gè)對(duì)象 sparkContext 是操作 RDD 的 sqlContext 是操作 sql 的

4、將數(shù)據(jù)源轉(zhuǎn)換為 DataSet/DataFrame

1、RDD
　　通過(guò)反射
　　通過(guò)自定義 schema 方式 2、通過(guò)使用 SparkSql 內(nèi)置數(shù)據(jù)源直接讀取 JSON、parquet、jdbc、orc、csv、text 文件，創(chuàng)建 DS/DF
　　hive 里用 orc 多
　　impla 里用 parquet 多

5、對(duì)4反射方式進(jìn)行解釋

　　同時(shí)，紅色字體處表示 import spark.implicits._ 排上用場(chǎng)

6、對(duì)4中通過(guò)自定義 schema 方式顯式的注入 schema 來(lái)生產(chǎn) DF

　　這個(gè) schema 由StructType 構(gòu)成，StructType 由StructFiledName，StructType，是否為空，這三部分組成

　　mode(SaveMode.override) 指的是，將數(shù)據(jù)寫(xiě)成文件時(shí)，如果存在這個(gè)目錄，則覆蓋掉

7、對(duì)4中，直接從數(shù)據(jù)源讀取數(shù)據(jù)，轉(zhuǎn)換成 DF 進(jìn)行解釋

　　這些數(shù)據(jù)源，內(nèi)部本身就包含了數(shù)據(jù)的 schema，所以可以直接讀取文件成一個(gè) DF

　　2是1的簡(jiǎn)寫(xiě)，區(qū)別是，如果是內(nèi)部數(shù)據(jù)源，用2，如果是外部數(shù)據(jù)源，用1

　　3是直接通過(guò) sql 的方式去創(chuàng)建成表，然后通過(guò) select 的方式去查找，然而編程的時(shí)候不是用這樣的寫(xiě)法，由其他 sql 寫(xiě)法

　　json 和 parquet 兩者方式一樣

8、讀取 JDBC 數(shù)據(jù)，產(chǎn)生 DF

9、通過(guò)讀取 text 來(lái)生成 DF

　　注意1與2的不同！

?10、引用外部數(shù)據(jù)源的方法

　　去下面網(wǎng)址，進(jìn)入 DataSource，查找外部數(shù)據(jù)源的使用

spark-packages.org

?11、RDD、DF、DS 之間的關(guān)系

　　首先從 DataSource 那里獲取數(shù)據(jù)，生成 DF，

　　　　DF 通過(guò).rdd 生成 RDD

　　　　DF 通過(guò).toDS 或者 .as 生成 DS

　　DF和 DS 都可以轉(zhuǎn)換成 RDD ，需要注意的是從 DF 轉(zhuǎn)換成 RDD 格式是 Row 對(duì)象，并不是它原始對(duì)象，DS 轉(zhuǎn)換成 RDD 格式可以是原始的對(duì)象

　　還可以把一個(gè) Scala 集合轉(zhuǎn)化成 DS，跟把 Scala 集合轉(zhuǎn)化成 RDD 一樣

　　圖有點(diǎn)老，在 spark2.1中，RDD.toDS 方法已經(jīng)有了

12、加載文件

1： scala> val usersRdd=sc.textFile("/Users/orco/data/ml-1m/users.dat") usersRdd: org.apache.spark.rdd.RDD[String] = /Users/orco/data/ml-1m/users.dat MapPartitionsRDD[1] at textFile at <console>:242： //json、orc、parquet、csv 讀取方式一樣，下面舉例兩個(gè) scala> val userJsonDF=spark.read.format("json").load("/tmp/user.json") userJsonDF: org.apache.spark.sql.DataFrame = [age: bigint, gender: string ... 3 more fields]//該讀取方式是上面方式的簡(jiǎn)寫(xiě)，內(nèi)部數(shù)據(jù)用下面的，外部數(shù)據(jù)用上面的 scala> val userParquetDF=spark.read.parquet("/tmp/user.parquet") userParquetDF: org.apache.spark.sql.DataFrame = [userID: bigint, gender: string ... 3 more fields]3： //spark.read.text 返回 DataFrame scala> val rdd = spark.read.text("/Users/orco/data/ml-1m/users.dat") rdd: org.apache.spark.sql.DataFrame = [value: string]//spark.read.textFile 返回 DataSet scala> val rdd = spark.read.textFile("/Users/orco/data/ml-1m/users.dat") rdd: org.apache.spark.sql.Dataset[String] = [value: string]

?13、DS to DF?

　　toDF()，可以加參數(shù)，每一列給定義一個(gè)名字

14、練習(xí)

1. json 數(shù)據(jù) {"age":"45","gender":"M","occupation":"7","userID":"4","zipcode":"02460"}{"age":"1","gend er":"F","occupation":"10","userID":"1","zipcode":"48067"} 2. 讀取數(shù)據(jù) scala> val userDF = spark.read.json("/tmp/user.json") userDF: org.apache.spark.sql.DataFrame = [age: string, gender: string, occupation: string, userID: string, zipcode: string] 3. 生成Json數(shù)據(jù) scala> userDF.limit(5).write.mode("overwrite").json("/tmp/user2.json") 4. 查看數(shù)據(jù) scala> userDF.show(4)或者(DF.toJSON 生成一個(gè) DS) scala> userDF.limit(2).toJSON.foreach(x =>println(x)) {"age":"1","gender":"F","occupation":"10","userID":"1","zipcode":"48067"} {"age":"56","gender":"M","occupation":"16","userID":"2","zipcode":"70072"}或者 scala> userDF.printSchema root|-- age: string (nullable = true)|-- gender: string (nullable = true)|-- occupation: string (nullable = true) |-- userID: string (nullable = true)|-- zipcode: string (nullable = true) 5. 修改 DF/DS 元信息 userDF.toDF("a","b","c","d","e") userDS.toDF("a","b","c","d","e")或者 val userDs = spark.read.textFile("ml-1m/users.dat").map(_.split("::")) val userDf = userDs.map(x => (x(0).toLong, x(1).toString, x(2).toInt, x(3).toInt, x(4))).toDF("userId", "gender", "age", "occ", "timestamp")或者 //增加新列“age2” userDf.withColumn(”age2",col(”age")+1) 6. Action 算子，如 collect、first、take、head 等

15、單獨(dú)列舉出來(lái)，select 算子

16、filter 的兩種使用

17、混用 select filter,無(wú)先后順序

userDF.select("userID", "age").filter("age > 30").show(2)userDF.filter("age > 30").select("userID", "age").show(2)

18、groupBy

scala> userDF.groupBy("age").count().show() +---+-----+ |age|count| +---+-----+ | 50| 496| | 25| 2096| | 56| 380| | 1| 222| | 35| 1193| | 18| 1103| | 45| 550| +---+-----+scala> userDF.groupBy("age").agg(count("gender")).show() +---+-------------+ |age|count(gender)| +---+-------------+ | 50| 496| | 25| 2096| | 56| 380| | 1| 222| | 35| 1193| | 18| 1103| | 45| 550| +---+-------------+scala> userDF.groupBy("age").agg(countDistinct("gender")).show() +---+----------------------+ |age|count(DISTINCT gender)| +---+----------------------+ | 50| 2| | 25| 2| | 56| 2| | 1| 2| | 35| 2| | 18| 2| | 45| 2| +---+----------------------+scala>

19、groupBy，agg 另一種寫(xiě)法

　　可用的聚集函數(shù):

　　`avg`, `max`, `min`, `sum`, `count`?

scala> userDF.groupBy("age").agg("gender"->"count","occupation"->"count").show() +---+-------------+-----------------+ |age|count(gender)|count(occupation)| +---+-------------+-----------------+ | 50| 496| 496| | 25| 2096| 2096| | 56| 380| 380| | 1| 222| 222| | 35| 1193| 1193| | 18| 1103| 1103| | 45| 550| 550| +---+-------------+-----------------+　

20、join

當(dāng) join 的列名不一樣的時(shí)候用下面的方式，同時(shí)可以指定連接方式，如 inner

21、DF 創(chuàng)建臨時(shí)表

　　不論是臨時(shí)表還是全局表，application 關(guān)閉后，都會(huì)刪除，如果想一直有效，那就用 saveAsTable 的方式存起來(lái)

userDataFrame.createOrReplaceTempView("users") val groupedUsers = spark.sql("select gender, age, count(*) as n from users group by gender, age") groupedUsers.show()

?

?

22、SparkSQL 的萬(wàn)能思路

第一步:得到DataFrame或Dataset val ds = ... 第二步:注冊(cè)成臨時(shí)表 ds.registerTempTable("xxx") 第三步:用SQL計(jì)算 spark.sql ("SELECT ...")

?

?

轉(zhuǎn)載于:https://www.cnblogs.com/sorco/p/7436480.html

與50位技術(shù)專(zhuān)家面對(duì)面20年技術(shù)見(jiàn)證，附贈(zèng)技術(shù)全景圖

總結(jié)

以上是生活随笔為你收集整理的SparkSQL程序设计的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。