?1、創建Spark Session

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

2、將RDD隱式轉換為DataFrame

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

4、將數據源轉換為 DataSet/DataFrame

1、RDD
　　通過反射
　　通過自定義 schema 方式 2、通過使用 SparkSql 內置數據源直接讀取 JSON、parquet、jdbc、orc、csv、text 文件，創建 DS/DF
　　hive 里用 orc 多
　　impla 里用 parquet 多

5、對4反射方式進行解釋

　　同時，紅色字體處表示 import spark.implicits._ 排上用場

6、對4中通過自定義 schema 方式顯式的注入 schema 來生產 DF

　　這個 schema 由StructType 構成，StructType 由StructFiledName，StructType，是否為空，這三部分組成

　　mode(SaveMode.override) 指的是，將數據寫成文件時，如果存在這個目錄，則覆蓋掉

7、對4中，直接從數據源讀取數據，轉換成 DF 進行解釋

　　這些數據源，內部本身就包含了數據的 schema，所以可以直接讀取文件成一個 DF

　　2是1的簡寫，區別是，如果是內部數據源，用2，如果是外部數據源，用1

　　3是直接通過 sql 的方式去創建成表，然后通過 select 的方式去查找，然而編程的時候不是用這樣的寫法，由其他 sql 寫法

　　json 和 parquet 兩者方式一樣

8、讀取 JDBC 數據，產生 DF

9、通過讀取 text 來生成 DF

　　注意1與2的不同！

?10、引用外部數據源的方法

　　去下面網址，進入 DataSource，查找外部數據源的使用

spark-packages.org

?11、RDD、DF、DS 之間的關系

　　首先從 DataSource 那里獲取數據，生成 DF，

　　　　DF 通過.rdd 生成 RDD

　　　　DF 通過.toDS 或者 .as 生成 DS

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

　　還可以把一個 Scala 集合轉化成 DS，跟把 Scala 集合轉化成 RDD 一樣

　　圖有點老，在 spark2.1中，RDD.toDS 方法已經有了

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 讀取方式一樣，下面舉例兩個 scala> val userJsonDF=spark.read.format("json").load("/tmp/user.json") userJsonDF: org.apache.spark.sql.DataFrame = [age: bigint, gender: string ... 3 more fields]//該讀取方式是上面方式的簡寫，內部數據用下面的，外部數據用上面的 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()，可以加參數，每一列給定義一個名字

14、練習

1. json 數據 {"age":"45","gender":"M","occupation":"7","userID":"4","zipcode":"02460"}{"age":"1","gend er":"F","occupation":"10","userID":"1","zipcode":"48067"} 2. 讀取數據 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數據 scala> userDF.limit(5).write.mode("overwrite").json("/tmp/user2.json") 4. 查看數據 scala> userDF.show(4)或者(DF.toJSON 生成一個 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、單獨列舉出來，select 算子

16、filter 的兩種使用

17、混用 select filter,無先后順序

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 另一種寫法

　　可用的聚集函數:

　　`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

當 join 的列名不一樣的時候用下面的方式，同時可以指定連接方式，如 inner

21、DF 創建臨時表

　　不論是臨時表還是全局表，application 關閉后，都會刪除，如果想一直有效，那就用 saveAsTable 的方式存起來

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

?

?

22、SparkSQL 的萬能思路

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

?

?

轉載于:https://www.cnblogs.com/sorco/p/7436480.html

與50位技術專家面對面20年技術見證，附贈技術全景圖

總結

以上是生活随笔為你收集整理的SparkSQL程序设计的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。