代碼與課件第五節(jié)課

1、第五節(jié)課SparkSQL歷史在hadoop發(fā)展過(guò)程中，為了給不理解MapReduce的技術(shù)提供快速上手的SQL工具，SQL on Hadoop的工具h(yuǎn)ive應(yīng)運(yùn)而生。但是，Hive中間環(huán)節(jié)消耗了大量的I/O，降低的運(yùn)行效率，進(jìn)而演化出MapR的DrillCloudera的Impala SharkShark是伯克利spark環(huán)境的組件之一，它修改了內(nèi)存管理、物理計(jì)劃、執(zhí)行三個(gè)模塊，并使之能運(yùn)行在spark引擎上，從而使得SQL的速度得到 0 00倍的提升；因?yàn)镾hark依賴(lài)hive太 ，制約spark各模塊的集合，拋棄shark代碼后，發(fā)展出sparkSQL；SparkSQL是分布式的即時(shí)工具，

2、屬于SPark系列的一個(gè)模塊，它利用了數(shù)據(jù)的結(jié)構(gòu)化信息，加入了的優(yōu)化措施，可以對(duì)結(jié)構(gòu)化數(shù)據(jù)進(jìn)行選擇、過(guò)濾、聚合等操作。SparkSQL優(yōu)點(diǎn)性能優(yōu)化方面 除了采取In Memory Columnar Storage、byte code generation等優(yōu)化技術(shù)外、將會(huì)引進(jìn)Cost M估、獲取最佳物理計(jì)劃等等對(duì)進(jìn)行動(dòng)態(tài)評(píng)組件擴(kuò)展方面 無(wú)論是SQL的語(yǔ)法器、分析器還是優(yōu)化器都可以 新定義，進(jìn)行擴(kuò)展Scala編寫(xiě)代碼的時(shí)候，盡量避免低效的、容易GC的代碼數(shù)據(jù)源兼容豐富：結(jié)構(gòu)化數(shù)據(jù)文件、Hive數(shù)據(jù)表、已的RDD、外部數(shù)據(jù)庫(kù)SparkSQL執(zhí)行與APISparkSQL優(yōu)化spark.sql.cod

3、egen 當(dāng)它設(shè)置為true時(shí)，Spark SQL會(huì)把每條的語(yǔ)句在運(yùn)行時(shí)編譯為java的二進(jìn)制代碼，大型時(shí)才開(kāi)啟Caching Data In Memory Then compression to minimize memory usage and GC pressurespark.sql.inMemoryColumnSpressed 自動(dòng)對(duì)內(nèi)存中的列式進(jìn)行壓縮spark.sql.inMemoryColumnarStorage.batchSize 列式緩存時(shí)的每個(gè)批處理的大小；增大可以提高內(nèi)存使用率與壓縮，但容易OOMpressed.codec 壓縮算法snappy/gzip/lzo spar

4、k.sql.tungsten.enabled tungsten優(yōu)化代碼：/* Created by ding on 20 8/ /6*/import org apache spark SparkContext import org apache spark SparkConf import org apache spark rdd RDDimport org apache spark sql DataFrame, SQLContextimport org apache spark storage StorageLevelobject sql /模式類(lèi) 默認(rèn)序列化與支持模式匹配case clas

5、s(name:String, age:Int)def main(args: ArrayString) val conf = new SparkConf() setAppName("SQL") setMaster("local4") val sc = new SparkContext(conf)sc setLogLevel("WARN")val wordcount = sc textFile("C:UserszxyDesktopSpark txt") flatMap(_ split(" ") ma

6、p(_,) reduceByKey(_+_) persist(StorageLevel MEMORY_AND_DISK) wordcount collect() foreach(println)val sqlContext = new SQLContext(sc)/* val spark = SparkSession builder()appName("Spark SQL basic example") config("spark some config option", "some value") getOrCreate()*/im

7、port sqlContext implicits _/創(chuàng)建一個(gè)數(shù)據(jù)集val peopleDF: DataFrame = sqlContext read json("C:UserszxyDesktoppeople json")println("=數(shù)據(jù)集查看=")/數(shù)據(jù)集查看peopleDF show()/查看數(shù)據(jù)表的結(jié)構(gòu)schema peopleDF printSchema()/選擇列進(jìn)行查看peopleDF select("name") show()peopleDF select($"name", $"

8、age" + ) show()/過(guò)濾列進(jìn)行查看peopleDF filter($"age" > 2 ) show()/聚合列peopleDF groupBy("age") count() show()/peopleDF createOrReplaceTempView("people") peopleDF registerTempTable("peopleTable") peopleDF registerTempTable("peopleTable_back")sqlContext

9、 sql("select a *, b age from peopleTable a join peopleTable_back b on a name = b name") showpeopleDF join(peopleDF) showprintln("=程序化/程序式的運(yùn)行SQL=")val temporaryPeopleDF : DataFrame = peopleDF select("name")temporaryPeopleDF show()val temporaryPeopleDF2: DataFrame = sqlCo

10、ntext sql("select age from peopleTable") temporaryPeopleDF2 show()sqlContext sql("select max(age) as maxAge from peopleTable") show peopleDF map(_ getAsString("name") foreach(println)peopleDF map(_ getValuesMapAny(List("name","age") foreach(println)p

11、rintln("=反射生成=")/DataFrame與RDD的互相轉(zhuǎn)換/ 使用反射推斷schema/代碼更加簡(jiǎn)潔 并且你已經(jīng)確定RDD的結(jié)構(gòu)import sqlContext implicits _val people = sc makeRDD(Seq("july,35", "tine, 8") map(_ split(",")val peopleDF_ : RDD = people map(p => peopleDF_ toDF() show()sqlContext createDataFrame(peop

12、leDF_ ) show()(p(0),p( ) trim toInt)println("=顯式指明=")/2 程序式的指明schema結(jié)構(gòu)/代碼復(fù)雜，數(shù)據(jù)運(yùn)行時(shí)才能確定結(jié)構(gòu)val schemaString = "name age"import org apache spark sql Rowimport org apache spark sql types StructType,StructField,StringType; val schema: StructType = StructType(schemaString split(" ") map(p=> StructField(p , StringType, true)val rowRD