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

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

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

3、egen 當它設置為true時，Spark SQL會把每條的語句在運行時編譯為java的二進制代碼，大型時才開啟Caching Data In Memory Then compression to minimize memory usage and GC pressurespark.sql.inMemoryColumnSpressed 自動對內(nèi)存中的列式進行壓縮spark.sql.inMemoryColumnarStorage.batchSize 列式緩存時的每個批處理的大??；增大可以提高內(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 /模式類 默認序列化與支持模式匹配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)建一個數(shù)據(jù)集val peopleDF: DataFrame = sqlContext read json("C:UserszxyDesktoppeople json")println("=數(shù)據(jù)集查看=")/數(shù)據(jù)集查看peopleDF show()/查看數(shù)據(jù)表的結構schema peopleDF printSchema()/選擇列進行查看peopleDF select("name") show()peopleDF select($"name", $"

8、age" + ) show()/過濾列進行查看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("=程序化/程序式的運行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的互相轉換/ 使用反射推斷schema/代碼更加簡潔 并且你已經(jīng)確定RDD的結構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結構/代碼復雜，數(shù)據(jù)運行時才能確定結構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