您好,登錄后才能下訂單哦!
這篇文章主要為大家分析了如何進(jìn)行Spark SQL中的Structured API分析的相關(guān)知識(shí)點(diǎn),內(nèi)容詳細(xì)易懂,操作細(xì)節(jié)合理,具有一定參考價(jià)值。如果感興趣的話,不妨跟著跟隨小編一起來(lái)看看,下面跟著小編一起深入學(xué)習(xí)“如何進(jìn)行Spark SQL中的Structured API分析”的知識(shí)吧。
Spark 中所有功能的入口點(diǎn)是 SparkSession
,可以使用 SparkSession.builder()
創(chuàng)建。創(chuàng)建后應(yīng)用程序就可以從現(xiàn)有 RDD,Hive 表或 Spark 數(shù)據(jù)源創(chuàng)建 DataFrame。示例如下:
val spark = SparkSession.builder().appName("Spark-SQL").master("local[2]").getOrCreate() val df = spark.read.json("/usr/file/json/emp.json") df.show() // 建議在進(jìn)行 spark SQL 編程前導(dǎo)入下面的隱式轉(zhuǎn)換,因?yàn)?nbsp;DataFrames 和 dataSets 中很多操作都依賴了隱式轉(zhuǎn)換 import spark.implicits._
可以使用 spark-shell
進(jìn)行測(cè)試,需要注意的是 spark-shell
啟動(dòng)后會(huì)自動(dòng)創(chuàng)建一個(gè)名為 spark
的 SparkSession
,在命令行中可以直接引用即可。
Spark 支持由內(nèi)部數(shù)據(jù)集和外部數(shù)據(jù)集來(lái)創(chuàng)建 DataSet,其創(chuàng)建方式分別如下:
// 1.需要導(dǎo)入隱式轉(zhuǎn)換 import spark.implicits._ // 2.創(chuàng)建 case class,等價(jià)于 Java Bean case class Emp(ename: String, comm: Double, deptno: Long, empno: Long, hiredate: String, job: String, mgr: Long, sal: Double) // 3.由外部數(shù)據(jù)集創(chuàng)建 Datasets val ds = spark.read.json("/usr/file/emp.json").as[Emp] ds.show()
// 1.需要導(dǎo)入隱式轉(zhuǎn)換 import spark.implicits._ // 2.創(chuàng)建 case class,等價(jià)于 Java Bean case class Emp(ename: String, comm: Double, deptno: Long, empno: Long, hiredate: String, job: String, mgr: Long, sal: Double) // 3.由內(nèi)部數(shù)據(jù)集創(chuàng)建 Datasets val caseClassDS = Seq(Emp("ALLEN", 300.0, 30, 7499, "1981-02-20 00:00:00", "SALESMAN", 7698, 1600.0), Emp("JONES", 300.0, 30, 7499, "1981-02-20 00:00:00", "SALESMAN", 7698, 1600.0)) .toDS() caseClassDS.show()
<br/>
Spark 支持兩種方式把 RDD 轉(zhuǎn)換為 DataFrame,分別是使用反射推斷和指定 Schema 轉(zhuǎn)換:
// 1.導(dǎo)入隱式轉(zhuǎn)換 import spark.implicits._ // 2.創(chuàng)建部門類 case class Dept(deptno: Long, dname: String, loc: String) // 3.創(chuàng)建 RDD 并轉(zhuǎn)換為 dataSet val rddToDS = spark.sparkContext .textFile("/usr/file/dept.txt") .map(_.split("\t")) .map(line => Dept(line(0).trim.toLong, line(1), line(2))) .toDS() // 如果調(diào)用 toDF() 則轉(zhuǎn)換為 dataFrame
import org.apache.spark.sql.Row import org.apache.spark.sql.types._ // 1.定義每個(gè)列的列類型 val fields = Array(StructField("deptno", LongType, nullable = true), StructField("dname", StringType, nullable = true), StructField("loc", StringType, nullable = true)) // 2.創(chuàng)建 schema val schema = StructType(fields) // 3.創(chuàng)建 RDD val deptRDD = spark.sparkContext.textFile("/usr/file/dept.txt") val rowRDD = deptRDD.map(_.split("\t")).map(line => Row(line(0).toLong, line(1), line(2))) // 4.將 RDD 轉(zhuǎn)換為 dataFrame val deptDF = spark.createDataFrame(rowRDD, schema) deptDF.show()
<br/>
Spark 提供了非常簡(jiǎn)單的轉(zhuǎn)換方法用于 DataFrame 與 Dataset 間的互相轉(zhuǎn)換,示例如下:
# DataFrames轉(zhuǎn)Datasets scala> df.as[Emp] res1: org.apache.spark.sql.Dataset[Emp] = [COMM: double, DEPTNO: bigint ... 6 more fields] # Datasets轉(zhuǎn)DataFrames scala> ds.toDF() res2: org.apache.spark.sql.DataFrame = [COMM: double, DEPTNO: bigint ... 6 more fields]
<br/>
Spark 支持多種方法來(lái)構(gòu)造和引用列,最簡(jiǎn)單的是使用 col()
或 column()
函數(shù)。
col("colName") column("colName") // 對(duì)于 Scala 語(yǔ)言而言,還可以使用$"myColumn"和'myColumn 這兩種語(yǔ)法糖進(jìn)行引用。 df.select($"ename", $"job").show() df.select('ename, 'job).show()
// 基于已有列值新增列 df.withColumn("upSal",$"sal"+1000) // 基于固定值新增列 df.withColumn("intCol",lit(1000))
// 支持刪除多個(gè)列 df.drop("comm","job").show()
df.withColumnRenamed("comm", "common").show()
需要說(shuō)明的是新增,刪除,重命名列都會(huì)產(chǎn)生新的 DataFrame,原來(lái)的 DataFrame 不會(huì)被改變。
<br/>
// 1.查詢員工姓名及工作 df.select($"ename", $"job").show() // 2.filter 查詢工資大于 2000 的員工信息 df.filter($"sal" > 2000).show() // 3.orderBy 按照部門編號(hào)降序,工資升序進(jìn)行查詢 df.orderBy(desc("deptno"), asc("sal")).show() // 4.limit 查詢工資最高的 3 名員工的信息 df.orderBy(desc("sal")).limit(3).show() // 5.distinct 查詢所有部門編號(hào) df.select("deptno").distinct().show() // 6.groupBy 分組統(tǒng)計(jì)部門人數(shù) df.groupBy("deptno").count().show()
<br/>
// 1.首先需要將 DataFrame 注冊(cè)為臨時(shí)視圖 df.createOrReplaceTempView("emp") // 2.查詢員工姓名及工作 spark.sql("SELECT ename,job FROM emp").show() // 3.查詢工資大于 2000 的員工信息 spark.sql("SELECT * FROM emp where sal > 2000").show() // 4.orderBy 按照部門編號(hào)降序,工資升序進(jìn)行查詢 spark.sql("SELECT * FROM emp ORDER BY deptno DESC,sal ASC").show() // 5.limit 查詢工資最高的 3 名員工的信息 spark.sql("SELECT * FROM emp ORDER BY sal DESC LIMIT 3").show() // 6.distinct 查詢所有部門編號(hào) spark.sql("SELECT DISTINCT(deptno) FROM emp").show() // 7.分組統(tǒng)計(jì)部門人數(shù) spark.sql("SELECT deptno,count(ename) FROM emp group by deptno").show()
上面使用 createOrReplaceTempView
創(chuàng)建的是會(huì)話臨時(shí)視圖,它的生命周期僅限于會(huì)話范圍,會(huì)隨會(huì)話的結(jié)束而結(jié)束。
你也可以使用 createGlobalTempView
創(chuàng)建全局臨時(shí)視圖,全局臨時(shí)視圖可以在所有會(huì)話之間共享,并直到整個(gè) Spark 應(yīng)用程序終止后才會(huì)消失。全局臨時(shí)視圖被定義在內(nèi)置的 global_temp
數(shù)據(jù)庫(kù)下,需要使用限定名稱進(jìn)行引用,如 SELECT * FROM global_temp.view1
。
// 注冊(cè)為全局臨時(shí)視圖 df.createGlobalTempView("gemp") // 使用限定名稱進(jìn)行引用 spark.sql("SELECT ename,job FROM global_temp.gemp").show()
這篇文章主要為大家分析了如何進(jìn)行Spark SQL中的Structured API分析的相關(guān)知識(shí)點(diǎn),內(nèi)容詳細(xì)易懂,操作細(xì)節(jié)合理,具有一定參考價(jià)值。如果感興趣的話,不妨跟著跟隨小編一起來(lái)看看,下面跟著小編一起深入學(xué)習(xí)“如何進(jìn)行Spark SQL中的Structured API分析”的知識(shí)吧。
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如果涉及侵權(quán)請(qǐng)聯(lián)系站長(zhǎng)郵箱:is@yisu.com進(jìn)行舉報(bào),并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。