大數(shù)據(jù)分析與實戰(zhàn)項目5SparkSQL處理健康監(jiān)控數(shù)據(jù)現(xiàn)有一組透析治療患者的健康監(jiān)測數(shù)據(jù)，要求針對性開展數(shù)據(jù)分析，為病患管理、健康監(jiān)控提供信息支持。健康醫(yī)療是廣大群眾關(guān)心的熱點問題，也是關(guān)系到社會穩(wěn)定、國家發(fā)展的重大課題。情境導(dǎo)入Spark項目分解Spark序號任務(wù)任務(wù)說明1認識SparkSQL及其數(shù)據(jù)結(jié)構(gòu)認識新的數(shù)據(jù)結(jié)構(gòu)DataFrame、DataSet，在SparkShell下查看血透病患的健康監(jiān)測數(shù)據(jù)。2由健康監(jiān)測數(shù)據(jù)創(chuàng)建DataFrame根據(jù)血液透析患者的健康監(jiān)測數(shù)據(jù)集，創(chuàng)建DataFrame，并打印相關(guān)信息。3DSL方式分析健康監(jiān)測數(shù)據(jù)分析以下指標：（1）透析10年以上患者的信息；（2）分組統(tǒng)計各基礎(chǔ)疾病患者的數(shù)量；（3）該組數(shù)據(jù)中，血壓偏高者的占比。4SQL方式分析健康監(jiān)測數(shù)據(jù)分析以下指標：（1）打印年齡在70歲（含）以上、有“慢性腎小球炎”基礎(chǔ)病的患者信息；（2）該組數(shù)據(jù)中，BMI偏高（肥胖）、BMI偏低（消瘦）人員的數(shù)量。5將梳理后的數(shù)據(jù)寫入MySQL數(shù)據(jù)庫表中抽取“慢性腎小球炎”患者的部分數(shù)據(jù)，寫入到MySQL數(shù)據(jù)庫表中。能使用DSL風(fēng)格或SQL風(fēng)格編程，完成DataFrame數(shù)據(jù)的分析。能否根據(jù)各種數(shù)據(jù)源，創(chuàng)建DataFrame。能將DataFrame數(shù)據(jù)分析的結(jié)果寫入數(shù)據(jù)庫或文件中。123學(xué)習(xí)目標Spark項目5

SparkSQL處理健康監(jiān)控數(shù)據(jù)Spark任務(wù)1認識SparkSQL及其數(shù)據(jù)抽象查看健康監(jiān)測數(shù)據(jù)DSL方式分析健康監(jiān)測數(shù)據(jù)任務(wù)2任務(wù)3SQL方式分析健康監(jiān)測數(shù)據(jù)任務(wù)4DataFrame數(shù)據(jù)寫入MySQL任務(wù)5任務(wù)分析Spark根據(jù)內(nèi)存數(shù)據(jù)，創(chuàng)建數(shù)據(jù)集DataFrame/Dataset，并查看其中的數(shù)據(jù)。SparkSQL的產(chǎn)生Spark產(chǎn)生根源：（1）MapReduce編寫程序比較復(fù)雜，需要大量學(xué)習(xí)；（2）Hive雖然支持類似SQL的查詢，但它運行在MapReduce之上，運算效率低。伯克利實驗室開發(fā)了基于Hive的結(jié)構(gòu)化數(shù)據(jù)處理組件Shark（SparkSQL的前身），但Shark繼承了大量的Hive代碼，后續(xù)優(yōu)化、維護較為麻煩。。SparkSQL是Spark體系中專門處理結(jié)構(gòu)化數(shù)據(jù)（StructuredData）的模塊，目前使用DataFrame、Dataset作為數(shù)據(jù)抽象。Spark的數(shù)據(jù)抽象SparkDataset中的數(shù)據(jù)形態(tài)下圖反映了RDD、DataFrame、Dataset的區(qū)別。注意：從Spark2.X開始，DataSet、DataFrame的API方法已做了統(tǒng)一，極大減輕了學(xué)習(xí)負擔。初步體驗SparkSQLSpark在SparkRDD學(xué)習(xí)的過程中可知，SparkContext（sc）是RDD開發(fā)的程序入口（上下文環(huán)境）；而SparkSQL也有自己的入口SparkSession。初步體驗SparkSQLSparkSparkSession提供了createDataFrame方法，它可以將內(nèi)存中的有序集合構(gòu)造成DataFrame；scala>valdata=List(("Tom","male",20),("Jerry","male",18))

//定義一個列表scala>valdf=spark.createDataFrame(data)//根據(jù)data生成DataFramescala>df.show()//顯示DataFrame中的數(shù)據(jù)初步體驗SparkSQLSparkscala>valdf=spark.createDataFrame(data).toDF("name","gender","age")//創(chuàng)建DataFrame，并為各列命名scala>df.show(2)//顯示df的前2行相關(guān)知識小結(jié)SparkSparkSQL專門處理結(jié)構(gòu)化數(shù)據(jù)，數(shù)據(jù)抽象為DataFrame、Dataset。createDataFrame方法可以創(chuàng)建DataFrame對象；show方法顯示DataFrame中的數(shù)據(jù)；toDF方法，修改DataFrame列的名稱。根據(jù)知識儲備的相關(guān)知識，利用內(nèi)存數(shù)據(jù)（患者信息）創(chuàng)建DataFrame，并查看其中的內(nèi)容。任務(wù)實施項目5

SparkSQL處理健康監(jiān)控數(shù)據(jù)Spark任務(wù)1認識SparkSQL及其數(shù)據(jù)抽象查看健康監(jiān)測數(shù)據(jù)DSL方式分析健康監(jiān)測數(shù)據(jù)任務(wù)2任務(wù)3SQL方式分析健康監(jiān)測數(shù)據(jù)任務(wù)4DataFrame數(shù)據(jù)寫入MySQL任務(wù)5任務(wù)分析Spark要想借助SparkSQL模塊開展數(shù)據(jù)分析，首先需要有DataFrame。在IDEA下，根據(jù)血液透析患者的健康監(jiān)測數(shù)據(jù)集healthdata.csv，創(chuàng)建DataFrame，填充“基礎(chǔ)病”字段（列）的缺失值，并打印前5行數(shù)據(jù)。由數(shù)據(jù)文件創(chuàng)建RDDSpark通過createDataFrame方法創(chuàng)建DataFrame通常應(yīng)用于小規(guī)模實驗（學(xué)習(xí)），實際業(yè)務(wù)中很多數(shù)據(jù)是存儲在JSON、CSV、Parquet等文件中，SparkSQL提供了read方法，可以讀取這些文件，并生成DataFrame。例如：JSON文件scala>valstudent=spark.read.json(filePath)//生成DataFramescala>valstudent=spark.read.format("json").load(filePath)//與上一行代碼等效由數(shù)據(jù)文件創(chuàng)建RDDSpark逗號分隔值（Comma-SeparatedValues，CSV）是以純文本形式存儲數(shù)據(jù)的文件，其數(shù)據(jù)之間通常用逗號分割（也可以是其他字符），是數(shù)據(jù)交換、處理領(lǐng)域常見的格式?，F(xiàn)有一個CSV文件student.csv，記錄了若干學(xué)生的信息：由數(shù)據(jù)文件創(chuàng)建RDDSpark讀取CSV文件創(chuàng)建DataFrame時，需要通過option方法指定header、sep等信息。scala>valstudent=spark.read.option("header","true").option("sep",",").csv(path)scala>valstudent=spark.read.format(“csv”).option("header",true).option("sep",",").load(path)scala>student.show()查看DataFrame中的數(shù)據(jù)SparkRDD經(jīng)常用take、collect、first等行動操作查看數(shù)據(jù)，DataFrame同樣提供了若干類似的行動操作方法。重復(fù)值、缺失值的處理SparkSparkSQL中，經(jīng)常用na方法處置空值，其返回值類型為DataFrameNaFunctions；而DataFrameNaFunctions有drop、fill等方法具體處理空值。scala>studentDF2.na.drop()//刪除缺失值scala>studentDF2.na.drop(Array("gender","score"))//刪除gender/score缺失值scala>studentDF2.na.fill(Map(("gender","male"),("score",0)))//填充缺失值scala>studentDF2.distinct()//刪除重復(fù)行DataFrame數(shù)據(jù)保存到文件中SparkSparkSQL能夠讀取各種文件生成DataFrame，同樣DataFrame也可以根據(jù)需要輸出到JSON、CSV等文件中。scala>student.write.format("json").save(path+"/json")//保存到j(luò)son文件中scala>student.write.option("header","true").format("csv").save(path+"/csv")//保存到csv文件中SparkSQL提供了read方法，讀取各種文件，生成DataFrame；查看DataFrame數(shù)據(jù)：show、first、head、collect等；缺失值、重復(fù)值：na、fill、drop、distinct等；save方法將DataFrame數(shù)據(jù)保存到文件中。Spark綜合利用本任務(wù)中的知識儲備，讀取患者數(shù)據(jù)csv文件，完成缺失值的處理，并查看其中的數(shù)據(jù)。任務(wù)實施相關(guān)知識小結(jié)Spark項目5

SparkSQL處理健康監(jiān)控數(shù)據(jù)Spark任務(wù)1認識SparkSQL及其數(shù)據(jù)抽象查看健康監(jiān)測數(shù)據(jù)DSL方式分析健康監(jiān)測數(shù)據(jù)任務(wù)2任務(wù)3SQL方式分析健康監(jiān)測數(shù)據(jù)任務(wù)4DataFrame數(shù)據(jù)寫入MySQL任務(wù)5任務(wù)分析SparkSparkSQL進行數(shù)據(jù)分析，包含DSL（DomainSpecificLanguage）和SQL兩種語法風(fēng)格；其中DSL方式提供了上百個方法（類似于RDD中的操作、算子），允許開發(fā)者調(diào)用這些方法完成數(shù)據(jù)的處理。針對給定的數(shù)據(jù)集，采用DSL方式，使用DataFrame常見方法分析以下指標：（1）查找透析10年以上患者的信息；（2）分組統(tǒng)計各基礎(chǔ)疾病患者的數(shù)量；（3）計算該組患者中，血壓偏高者的占比。數(shù)據(jù)的查詢與篩選Spark在數(shù)據(jù)分析過程中，有時我們僅需要查詢某些列的值，可以借助select、selectExp等方法。例如，查詢studentDF中的name、age、score等3列，下列寫法等效：scala>valdf=studentDF.select("name","age","score").show(2)scala>valdf=studentDF.select(col("name"),col("age"),col("score")).show(2)

scala>valdf=studentDF.select($"name,$"age",$"score").show(2)

scala>valdf=studentDF.select(‘name,‘a(chǎn)ge,‘score).show(2)

數(shù)據(jù)的查詢與篩選Spark實際業(yè)務(wù)中，經(jīng)常需要根據(jù)某些條件過濾出某些行；SparkSQL提供了where與filter方法，要用于篩選出符合某條件的行，兩者用法一致。scala>valdf=studentDF.where("age>20")scala>valdf=

studentDF.where(col("age")>20)scala>valdf=

studentDF.where($"age">20and$"score">90)

數(shù)據(jù)的查詢與篩選Spark實際業(yè)務(wù)中，經(jīng)常需要根據(jù)某些條件過濾出某些行；SparkSQL提供了where與filter方法，要用于篩選出符合某條件的行，兩者用法一致。scala>valdf=studentDF.where("age>20")scala>valdf=

studentDF.where(col("age")>20)scala>valdf=

studentDF.where($"age">20and$"score">90)

數(shù)據(jù)的排序Spark輸出查詢結(jié)果前，經(jīng)常要進行排序，從而輸出前N行；SparkSQL提供了orderBy、sort方法用于排序，二者等效、用法一致。scala>valdf=studentDF.orderBy("score")

//按照score升序排列，得到新DFscala>studentDF.orderBy(col("score").desc)//按照score列的值降序排列DataFrame的連接join操作Sparkjoin操作用于連接兩個DataFrame組成一個新的DataFrame。scala>val

df1=spark.createDataFrame(List(("Tom",20),("Jerry",18),("Bob",19))).toDF("name","age")scala>valdf2=spark.createDataFrame(List(("Tom",176),("Jerry",182))).toDF("name","height")scala>df1.join(df2,"name",

"inner"

).show()//df1與df2做join操作，以name為連接字段scala>df1.join(df2,df1("name")===df2("name")).show()//與上一行等價，注意用三個等號與數(shù)據(jù)庫表的連接一致，DF也支持left（左）、right（右）、full（全）連接；只需替換join操作中的第3個參數(shù)字符串即可。聚合與分組統(tǒng)計Spark在開展數(shù)據(jù)分析的過程中，經(jīng)常用到各種聚合操作，比如求某列的和、平均值、最大值等，SparkSQL提供了相應(yīng)的方法。聚合與分組統(tǒng)計Sparkscala>studentDF.select(sum("score").alias("totalScore"),

avg("score").alias("avgScore")).show()聚合與分組統(tǒng)計Spark在數(shù)據(jù)統(tǒng)計的過程中，經(jīng)常需要分組統(tǒng)計；groupBy方法可以按照某個字段分組；groupBy通常與count、mean、max、min、sum等聚合操作聯(lián)合使用，從而完成各種分組統(tǒng)計任務(wù)scala>studentDF.groupBy("gender").agg(max("score"),min("score")).show()操作DataFrame的列SparkwithColumnRenamed重命名指定列（字段）的名稱。studentDF.withColumnRenamed("gender","sex")//gender列改名為sexwihtColumn方法可在當前DataFrame中增加一列。studentDF.withColumn("age+1",$"age"+1)//增加一個新列，其值為age+1查詢數(shù)據(jù)：select、where、filter等；DataFrame排序：sort或orderBy；join方法連接兩個DataFrame；統(tǒng)計分析方法：groupBy、agg、max、min、avg等Spark綜合利用本任務(wù)中的知識儲備，借助DataFrame的各種操作，完成指標的統(tǒng)計分析。任務(wù)實施相關(guān)知識小結(jié)Spark項目5

SparkSQL處理健康監(jiān)控數(shù)據(jù)Spark任務(wù)1認識SparkSQL及其數(shù)據(jù)抽象查看健康監(jiān)測數(shù)據(jù)DSL方式分析健康監(jiān)測數(shù)據(jù)任務(wù)2任務(wù)3SQL方式分析健康監(jiān)測數(shù)據(jù)任務(wù)4DataFrame數(shù)據(jù)寫入MySQL任務(wù)5任務(wù)分析Spark除了DSL方式，SparkSQL還提供了通過SQL語句方式處理數(shù)據(jù)。編寫SQL風(fēng)格的程序：（1）打印年齡在70歲（含）以上、有“慢性腎小球炎”基礎(chǔ)病的患者信息；（2）計算BMI，統(tǒng)計偏高（肥胖）、BMI偏低（消瘦）人員的數(shù)量。創(chuàng)建臨時視圖Spark在數(shù)據(jù)庫中，使用SQL語句進行查詢分析，需要有數(shù)據(jù)庫表（或者視圖）。用SQL方式處理DataFrame數(shù)據(jù)前，也需要將DataFrame對象注冊為一個臨時視圖（可以理解為虛擬表）。scala>studentDF.createTempView("student")//創(chuàng)建會話臨時視圖scala>studentDF.createOrReplaceTempView("student")//創(chuàng)建或者替換臨時視圖scala>studentDF.createGlobalTempView("student")//創(chuàng)建全局臨時視圖scala>studentDF.createOrReplaceGlobalTempView("student")//創(chuàng)建或者替換全局臨時視圖使用SQL風(fēng)格查詢分析Spark臨時視圖相當于數(shù)據(jù)庫中的表，臨時視圖的名字即相當于表的名字，在select子句中用臨時視圖名代替表名。scala>valsqlStr="selectname,age,genderfromstudentwhereage>20andgender='male'"http://SQL語句scala>valtempDF=spark.sql(sqlStr)//執(zhí)行SQL語句，返回一個DataFrame用戶自定義函數(shù)UDFSpark不少場景是內(nèi)置函數(shù)無法勝任的（或?qū)崿F(xiàn)較為復(fù)雜），這就需要開發(fā)人員編寫自定義函數(shù)（UserDefineFunction，簡稱UDF）來滿足需求。//定義一個匿名函數(shù)，并將其注冊為自定義函數(shù)（adjustAge）scala>spark.udf.register("adjustAge",(age:Int)=>if(age<18)"minor"else"adult")//在SQL語句中使用UDF函數(shù)adjustAgescala>valsqlStr="selectname,age,adjustAge(age)fromstudent"scala>spark.sql(sqlStr).show()//查詢、并顯示結(jié)果SQL方式：將DataFrame注冊為臨時視圖，書寫普通的SQL語句，然后執(zhí)行；當SQL內(nèi)置函數(shù)不能滿足需求時，書寫用戶自定義函數(shù)。Spark綜合利用本任務(wù)中的知識儲備，利用SQL方式，完成相關(guān)指標的統(tǒng)計分析。任務(wù)實施相關(guān)知識小結(jié)Spark項目5

SparkSQL處理健康監(jiān)控數(shù)據(jù)Spark任務(wù)1認識SparkSQL及其數(shù)據(jù)抽象查看健康監(jiān)測數(shù)據(jù)DSL方式分析健康監(jiān)測數(shù)據(jù)任務(wù)2任務(wù)3SQL方式分析健康監(jiān)測數(shù)據(jù)任務(wù)4DataFrame數(shù)據(jù)寫入MySQL任務(wù)5任務(wù)分析Spark借助JDBC連接器，SparkSQL可以實現(xiàn)與MySQL等數(shù)據(jù)庫的互聯(lián)互通，數(shù)據(jù)處理后的結(jié)果也可以寫入到MySQL中。計算每位患者的透析脫水量（透析前的體重grossweight減去體重weight，是透析治療特別關(guān)注的指標），按照脫水量排序，取Top5的病人信息，寫入到MySQL數(shù)據(jù)庫的表health中。MySQL相關(guān)準備工作Spark使用“sudoapt-getinstallmysql-server”命令，完成MySQL的安裝。alteruser'root'@'localhost'identifiedwithmysql_native_passwordby'123';#修改root密碼flushprivileges;#刷新權(quán)限createdatabasesparkTest;#創(chuàng)建數(shù)據(jù)庫sparkTestusesparkTest;#使用數(shù)據(jù)庫sparkTestcreatetablestudent(idchar(10),namechar(20),sexchar(10),ageint(4),addresschar(50));#創(chuàng)建表student讀取MySQL創(chuàng)建DataFrameSpark使用spark.read.format("jdbc")可以讀取數(shù)據(jù)庫表中的數(shù)據(jù)。scala>valstudentDF=spark.read.format("jdbc").option("driver","com.mysql.jdbc.Driver")//指定連接驅(qū)動名稱.option("url","jdbc:mysql://localhost:3306/sparkTest")//指定服務(wù)器.option("dbtable","student")//指定數(shù)據(jù)庫名稱.option("user","root")//指定連接的用戶.option("password","123")//指定用戶的密碼.load()DataFrame數(shù)據(jù)寫入MySQLSparkDataFrame