SparkSQL程序设计
?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兩個對象 sparkContext 是操作 RDD 的 sqlContext 是操作 sql 的4、將數(shù)據(jù)源轉(zhuǎn)換為 DataSet/DataFrame
1、RDD通過反射
通過自定義 schema 方式 2、通過使用 SparkSql 內(nèi)置數(shù)據(jù)源直接讀取 JSON、parquet、jdbc、orc、csv、text 文件，創(chuàng)建 DS/DF
hive 里用 orc 多
impla 里用 parquet 多
5、對4反射方式進行解釋
同時，紅色字體處表示 import spark.implicits._ 排上用場
6、對4中通過自定義 schema 方式顯式的注入 schema 來生產(chǎn) DF
這個 schema 由StructType 構(gòu)成，StructType 由StructFiledName，StructType，是否為空，這三部分組成
mode(SaveMode.override) 指的是，將數(shù)據(jù)寫成文件時，如果存在這個目錄，則覆蓋掉
7、對4中，直接從數(shù)據(jù)源讀取數(shù)據(jù)，轉(zhuǎn)換成 DF 進行解釋
這些數(shù)據(jù)源，內(nèi)部本身就包含了數(shù)據(jù)的 schema，所以可以直接讀取文件成一個 DF
2是1的簡寫，區(qū)別是，如果是內(nèi)部數(shù)據(jù)源，用2，如果是外部數(shù)據(jù)源，用1
3是直接通過 sql 的方式去創(chuàng)建成表，然后通過 select 的方式去查找，然而編程的時候不是用這樣的寫法，由其他 sql 寫法
json 和 parquet 兩者方式一樣
8、讀取 JDBC 數(shù)據(jù)，產(chǎn)生 DF
9、通過讀取 text 來生成 DF
注意1與2的不同！
?10、引用外部數(shù)據(jù)源的方法
去下面網(wǎng)址，進入 DataSource，查找外部數(shù)據(jù)源的使用
spark-packages.org?11、RDD、DF、DS 之間的關(guān)系
首先從 DataSource 那里獲取數(shù)據(jù)，生成 DF，
DF 通過.rdd 生成 RDD
DF 通過.toDS 或者 .as 生成 DS
DF和 DS 都可以轉(zhuǎn)換成 RDD ，需要注意的是從 DF 轉(zhuǎn)換成 RDD 格式是 Row 對象，并不是它原始對象，DS 轉(zhuǎn)換成 RDD 格式可以是原始的對象
還可以把一個 Scala 集合轉(zhuǎn)化成 DS，跟把 Scala 集合轉(zhuǎn)化成 RDD 一樣
圖有點老，在 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 讀取方式一樣，下面舉例兩個 scala> val userJsonDF=spark.read.format("json").load("/tmp/user.json") userJsonDF: org.apache.spark.sql.DataFrame = [age: bigint, gender: string ... 3 more fields]//該讀取方式是上面方式的簡寫，內(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ù)，每一列給定義一個名字
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 生成一個 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 另一種寫法
可用的聚集函數(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 的列名不一樣的時候用下面的方式，同時可以指定連接方式，如 inner
21、DF 創(chuàng)建臨時表
不論是臨時表還是全局表，application 關(guān)閉后，都會刪除，如果想一直有效，那就用 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 ...")?
?
轉(zhuǎn)載于:https://www.cnblogs.com/sorco/p/7436480.html
與50位技術(shù)專家面對面20年技術(shù)見證，附贈技術(shù)全景圖總結(jié)
以上是生活随笔為你收集整理的SparkSQL程序设计的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 路由配置
- 下一篇： OVS中arp响应的流表的实现