H13-723大數據HCIP-BigDataDeveloperV2.0中文V2.0汪孟德/wwx7118422019.06.30喻鳳/wx552338新開發陳星/cwx5259502020.11.30汪孟德/wwx711842賈云濤/jwx697590優化大數據離線批處理場景化解決方案進入大數據時代，企業產生的數據出現爆發式增長，部分數據需要實現離線存儲分析，而傳統的數據處理方案滿足不了海量數據存儲和海量數據處理需求。結合大數據離線技術，如何提出行之有效的解決方案以及如何去實施應用，成為企業面臨的難題。學完本課程后，您將能夠：熟悉離線批處理應用場景熟悉離線批處理采用的技術方案學習離線批處理的實際案例離線處理方案離線處理技術框架介紹數據存儲HDFS數據倉庫Hive離線分析SparkSQL數據采集工具離線批處理實戰業務場景-金融領域大數據平臺交易系統區數據服務區數據應用區數據倉庫系統監管報送類應用數據集成區外部數據人行征信社交網絡內部結構化數據核心信貸收單內部非結構化數據電話銀行網銀信用卡影像平臺電商平臺數據采集網絡爬蟲數據操控數據傳輸數據分發數據壓縮數據加載實時加載準實時加載操作型數據倉儲（準實時ODS）實驗數據區歷史數據區緩沖層基礎層匯總層公共實驗區分行實驗項目大數據存儲個人/企業征信銀監會1104報送證監會報送審計署平臺BI類應用零售分析系統電子銀行BI信用卡BI新型分析類應用精準營銷系統反欺詐/反洗錢個人信貸評分……KPI考核指標大數據多表關聯分析大數據算法分析離線分析平臺離線批處理概念離線批處理，是指對海量歷史數據進處理和分析，生成結果數據，供下一步數據應用使用的過程。離線批處理對數據處理的時延要求不高，但是處理的數據量較大，占用的計算存儲資源較多，通常通過MR作業、Spark作業或者HQL作業實現。核心訴求處理時間要求不高；處理數據量巨大；處理數據格式多樣；支持SQL類作業和自定義作業；離線批處理流程圖業務應用專題庫數據倉庫數據集市離線處理平臺數據源數據庫數據采集實時采集系統Flume/第三方采集等批量采集系統Flume/Sqoop/第三方ETL等離線批處理引擎MapReduce流式數據文件數據YarnHDFSSparkHiveSparkSQL集市應用倉庫應用（交互式分析）專題應用離線批處理常用組件HDFS：分布式文件系統，為各種批處理引擎提供數據存儲，可以存儲各種文件格式數據。YARN：資源調度引擎，為各種批處理引擎提供資源調度能力。MapReduce：大數據批處理引擎，用于處理海量數據，但是處理速度較慢。Hive：大數據SQL批處理引擎，用于處理SQL類批處理作業，但是處理速度較慢。Spark：基于內存的數據處理引擎，適合海量數據，處理速度高效。SparkSQL：Spark處理結構化數據的一個模塊。（多選題）以下哪些是離線批處理的特點？（）A.處理數據格式多樣B.支持SQL類作業和自定義作業C.處理數據量巨大D.處理時間要求高（單選題）離線批處理常用的組件不包括哪個？（）A.FlumeB.HiveC.SparkD.StormHive是大數據SQL批處理引擎，用于處理SQL類批處理作業，但是處理速度較慢。（）

離線處理方案離線處理技術框架介紹數據存儲HDFS數據倉庫Hive離線分析SparkSQL數據采集工具離線批處理實戰HDFS概述HDFS(HadoopDistributedFileSystem)基于Google發布的GFS論文設計開發。其除具備其它分布式文件系統相同特性外，HDFS還有自己特有的特性：高容錯性：認為硬件總是不可靠的。高吞吐量：為大量數據訪問的應用提供高吞吐量支持。大文件存儲：支持存儲TB-PB級別的數據。HDFS不適合做什么？大量小文件存儲隨機寫入低延遲讀取HDFS適合做什么？大文件存儲與訪問流式數據訪問基本系統架構HDFS架構圖Metadata(元數據):/home/foo/data,3,…DataNode機架

1機架

2DataNodesReplicationMetadataopsBlockopsNameNodeReadClientClientBlocksBlocks常用shell命令命令類別命令命令說明hdfsdfs-cat顯示文件內容-ls顯示目錄列表-rm刪除文件-put上傳目錄/文件到HDFS-get從HDFS下載目錄/文件到本地-mkdir創建目錄-chmod/-chown改變文件屬組……hdfsdfsadmin-safemode安全模式操作-report報告服務狀態HDFS回收站機制在HDFS里，刪除文件時，不會真正的刪除，其實是放入回收站，回收站里的文件可以用來快速恢復誤刪文件。可以設置一個時間閥值（單位：分鐘），當回收站里文件的存放時間超過這個閥值或是回收站被清空時，文件才會被徹底刪除，并且釋放占用的數據塊。Hadoop回收站trash，默認是關閉的，若開啟需要修改配置文件core-site.xml。<property>

<name>erval</name>

<value>1440</value></property>（多選題）HDFS里包含哪些實例？（）A.NameNodeB.TaskManagerC.DataNodeD.JobManager在HDFS根目錄下有一個文件a.txt，我們應該如何刪除？（判斷題）Hadoop回收站trash，若開啟需要修改配置文件core-site.xml。（）離線處理方案離線處理技術框架介紹數據存儲HDFS數據倉庫Hive離線分析SparkSQL數據采集工具離線批處理實戰Hive概述Hive是基于Hadoop的數據倉庫軟件，可以查詢和管理PB級別的分布式數據。Hive特性：靈活方便的ETL(Extract/Transform/Load)。支持MapReduce、Tez、Spark多種計算引擎。可直接訪問HDFS文件以及HBase。易用易編程。Hive的架構HiveDriver(Compiler,Optimizer,Executor)MetaStoreThriftServerWebInterfaceTezMapReduceSparkHive的數據存儲模型數據庫內部表外部表傾斜數據分區分區桶桶桶桶正常數據Hive內部表和外部表的區別區別：查詢表的類型：修改內部表tableName為外部表：修改外部表tableName為內部表：關鍵字內部表外部表CREATE/LOAD數據移到倉庫目錄數據位置不移動DROP元數據和數據會被一起刪除只刪除元數據descformattedtableName;altertabletableNamesettblproperties('EXTERNAL'='TRUE');altertabletableNamesettblproperties('EXTERNAL'='FALSE');Hive內置函數查看系統函數的用法：顯示函數的用法：詳細顯示函數的用法：常用函數：數學函數，如round()、abs()、rand()等。日期函數，如to_date()、current_date()等。字符串函數，如trim()、length()、substr()等。hive>showfunctions;hive>descfunctionupper;hive>descfunctionextendedupper;Hive自定義UDF當Hive提供的內置函數無法滿足業務處理需要時，此時就可以考慮使用用戶自定義函數，編寫處理代碼并在查詢中使用。UDF(User-Defined-Function）用于接收單個數據行，并產生一個數據行作為輸出。UDAF(User-DefinedAggregationFunction）用于接收多個數據行，并產生一個數據行作為輸出。UDTF(User-DefinedTable-GeneratingFunctions）用于接收單個數據行，并產生多個數據行作為輸出。UDF開發步驟繼承”org.apache.hadoop.hive.ql.exec.UDF”。實現一個evaluate()方法，編寫要實現的邏輯。打包并上傳到HDFS里。Hive創建臨時函數。調用該函數。Hive調優(1)數據傾斜數據傾斜指計算數據的時候，數據的分散度不夠，導致大量的數據集中到了一臺或者幾臺機器上計算，這些數據的計算速度遠遠低于平均計算速度，導致整個計算過程過慢。日常使用過程中，容易造成數據傾斜的原因可以歸納為如下幾點：group

bydistinct

count(distinct

xx)joinHive調優(2)調優參數：在map中會做部分聚集操作，效率更高但需要更多的內存。此時生成的查詢計劃會有兩個MRJob，可實現數據傾斜時負載均衡。set

hive.map.aggr=true；set

hive.groupby.skewindata=true;Hive調優(3)mapsidejoin當連接一個較小和較大表的時候，把較小的表直接放到內存中去，然后再對較大的表進行map操作。sethive.auto.convert.join=true;Hive調優(4)并行化執行每個查詢會被Hive轉化為多個階段，當有些階段關聯性不大時，可以并行化執行，減少整個任務的執行時間。開啟任務并行執行：設置同一個sql允許并行任務的最大線程數（例如設置為8個）：sethive.exec.parallel=true;sethive.exec.parallel.thread.number=8;HQL開發場景說明假定用戶開發一個基于Hive的數據分析應用，用于分析企業雇員信息。假定需要創建三張表：雇員信息表：“employees_info”。雇員聯絡信息表：“employees_concat”。雇員信息擴展表：”employees_info_extended”。雇員信息表編號姓名支付薪水幣種薪水金額繳稅稅種工作地入職時間1WangR8000.01personalincometax&0.05China:Shenzhen20193TomD12000.02personalincometax&0.09America:NewYork20194JackD24000.03personalincometax&0.09America:Manhattan20196LindaD36000.04personalincometax&0.09America:NewYork20198ZhangR48000.05personalincometax&0.05China:Shanghai2019雇員信息表設計字段字段名數據類型備注id編號INT編號name姓名STRING姓名usd_flag支付薪水幣種STRINGR是人民幣，D是美元salary薪水金額DOUBLE薪水金額deductions繳稅稅種MAP<STRING,DOUBLE>中間分隔符是&address工作地STRING工作地entrytime入職時間STRING入職時間雇員聯絡信息表編號電話號碼e-mail1135xxxxxxxxxxxx@3187xxxxxxxxxxxx@4182xxxxxxxxxxxx@6138xxxxxxxxxxxx@8134xxxxxxxxxxxx@雇員聯絡信息表設計字段字段名數據類型備注id編號INT編號name電話號碼STRING電話號碼e-mail郵箱STRING郵箱統計要求查看薪水支付幣種為美元的雇員聯系方式。查詢入職時間為2019年的雇員編號、姓名和電話號碼字段，并將查詢結果加載進表employees_info_extended對應的分區中。統計表employees_info中有多少條記錄。查詢以”cn”結尾的郵箱的員工信息。創建雇員信息表CREATEEXTERNALTABLEIFNOTEXISTSemployees_info(idINTComment"編號",nameSTRINGComment"姓名",usd_flagSTRINGComment"支付薪水幣種",salaryDOUBLEComment"薪水金額",deductionsMAP<STRING,DOUBLE>Comment"繳稅稅種",addressSTRINGComment"工作地",entrytimeSTRINGComment"入職時間")--指定行中各字段分隔符--"delimitedfieldsterminatedby"指定列與列之間的分隔符為','，"MAPKEYSTERMINATEDBY"指定MAP中鍵值的分隔符為'&'.ROWFORMATdelimitedfieldsterminatedby','MAPKEYSTERMINATEDBY'&'--指定表的存儲格式為TEXTFILE.STOREDASTEXTFILE;創建雇員聯絡信息表CREATEEXTERNALTABLEIFNOTEXISTSemployees_contact(idINTComment"編號",tel_phoneSTRINGComment"電話號碼",emailSTRINGComment"郵箱")--指定行中各字段分隔符.--"delimitedfieldsterminatedby"指定列與列之間的分隔符為','ROWFORMATdelimitedfieldsterminatedby','--指定表的存儲格式為TEXTFILE.STOREDASTEXTFILE;創建雇員信息擴展表CREATEEXTERNALTABLEIFNOTEXISTSemployees_info_extended(idINTComment"編號",nameSTRINGComment"姓名",usd_flagSTRINGComment"支付薪水幣種",salaryDOUBLEComment"薪水金額",deductionsMAP<STRING,DOUBLE>Comment"繳稅稅種",addressSTRINGComment“地址",tel_phoneSTRINGComment"電話號碼",emailSTRINGComment"郵箱")--使用關鍵字PARTITIONEDBY指定分區列名及數據類型.PARTITIONEDBY(entrytimeSTRING)STOREDASTEXTFILE;數據加載從本地加載數據到雇員信息表從HDFS加載數據到雇員聯絡信息表--從HDFS上/user/hive_examples_data/employees_contact.txt加載進employees_contact表中.LOADDATAINPATH'/user/hive_examples_data/employees_contact.txt'OVERWRITEINTOTABLEemployees_contact;--從本地文件系統/opt/hive_examples_data/目錄下將employee_info.txt加載進employees_info表中.LOADDATALOCALINPATH'/opt/hive_examples_data/employee_info.txt'OVERWRITEINTOTABLEemployees_info;查詢(1)查看薪水支付幣種為美元的雇員聯系方式。SELECT,b.tel_phone,b.emailFROMemployees_infoaJOINemployees_contactbONa.id=b.idWHEREusd_flag='D';查詢(2)查詢入職時間為2019年的雇員編號、姓名和電話號碼字段，并將查詢結果加載進表employees_info_extended中的入職時間為2019的分區中。INSERTOVERWRITETABLEemployees_info_extendedPARTITION(entrytime='2019')SELECTa.id,,b.tel_phoneFROMemployees_infoaJOINemployees_contactbONa.id=b.idWHEREa.entrytime='2019';查詢(3)使用Hive中自帶的函數COUNT()，統計表employees_concat中有多少條記錄。查詢以”cn”結尾的郵箱的員工信息。SELECTCOUNT(*)FROMemployees_contact;SELECT,b.tel_phoneFROMemployees_infoaJOINemployees_contactbONa.id=b.idWHEREb.emaillike'%cn';數據集市和數據倉庫的區別數據集市數據集市(DataMart)，也叫數據市場，數據集市就是滿足特定的部門或者用戶的需求，按照多維的方式進行存儲，包括定義維度、需要計算的指標、維度的層次等，生成面向決策分析需求的數據立方體。數據倉庫為滿足各類零散分析的需求，通過數據分層和數據模型的方式，并以基于業務和應用的角度將數據進行模塊化的存儲。Hive數據倉庫數據倉庫分層：ODS層：原始數據層。DWD層：結構和粒度與原始表保持一致，簡單清洗。DWS層：以DWD為基礎，進行輕度匯總。ADS層：為各種統計報表提供數據。ADS(ApplicationDataStore)DWS(DataWarehouseService)DWD(DataWarehouseDeatail)ODS(OperationalDataStore)分層的優點復雜問題簡單化將任務分解成多個步驟完成，每一層只處理單一的步驟，比較簡單，并且方便定位問題。減少重復開發規范數據分層，通過中間層數據，減少最大的重復計算，增加一次計算結果的復用性。隔離原始數據避免數據異常或者數據敏感，使真實數據與統計數據解耦。（多選題）Hive的自定義函數有哪幾種？

()UDFUDTFUDCFUDAF（判斷題）Hive在刪除表的時候，內部表的元數據和實際數據不會被一起刪除。

()（判斷題）Hive構建數據倉庫時通常需要進行分層。

()離線處理方案離線處理技術框架介紹數據存儲HDFS數據倉庫Hive離線分析SparkSQL數據采集工具離線批處理實戰Spark簡介Spark是基于內存的分布式批處理系統，它把任務拆分，然后分配到多個的CPU上進行處理，處理數據時產生的中間產物（計算結果）存放在內存中，減少了對磁盤的I/O操作，大大的提升了數據的處理速度，在數據處理和數據挖掘方面比較占優勢。Spark應用場景數據處理(DataProcessing)：可以用來快速處理數據，兼具容錯性和可擴展性。迭代計算(IterativeComputation)：支持迭代計算，有效應對復雜的數據處理邏輯。數據挖掘(DataMining)：在海量數據基礎上進行復雜的挖掘分析，可支持多種數據挖掘和機器學習算法。流式處理(StreamingProcessing)：支持秒級延遲的流處理，可支持多種外部數據源。查詢分析(QueryAnalysis)：支持SQL的查詢分析，同時提供領域特定語言(DSL)以方便操作結構化數據，并支持多種外部數據源。Spark對比MapReduce性能上提升了100倍。Spark的中間數據放在內存中，對于迭代運算的效率更高；進行批處理時更高效，同時有著更低的延遲。Spark提供更多的數據集操作類型，編程模型比MapReduce更靈活，開發效率更高。更高的容錯能力（血統機制）。RDDRDD是分布式彈性數據集，可以理解一個存儲數據的數據結構。Spark會把所要操作的數據，加載到RDD上，即RDD所有操作都是基于RDD來進行的。RDD是只讀和可分區。要想對RDD進行操作，只能重新生成一個新的RDD。從HDFS輸入創建，或從與Hadoop兼容的其他存儲系統中輸入創建。從父的RDD轉換的到新的RDD。從數據集合轉換而來，通過編碼實現。RDD的存儲：用戶可以選擇不同的存儲級別緩存RDD以便重用。當前RDD默認是存儲于內存，但當內存不足時，RDD會溢出到磁盤中。ShuffleShuffle是劃分DAG中stage的標識,同時影響Spark執行速度的關鍵步驟RDD的Transformation函數中,分為窄依賴(narrowdependency)和寬依賴(widedependency)的操作.窄依賴跟寬依賴的區別是是否發生Shuffle(洗牌)操作。窄依賴窄依賴是指父RDD的每個分區只被子RDD的一個分區所使用。表現為：一個父RDD的每一個分區對應于一個子RDD分區。unionmap,filterNarrowDependencies:joinwithinputsco-partitioned寬依賴寬依賴是指父RDD的每個分區都可能被多個子RDD分區所使用。表現為：父RDD的每個分區都被多個子RDD分區使用WideDependencies:groupByKeyjoinwithinputsnotco-partitionedStageA:B:C:D:Stage2E:unionStage3joinG:groupbyF:mapStage1TransformationTransformation是RDD的算子類型，它的返回值還是一個RDD。Transformation操作屬于懶操作（算子），不會真正觸發RDD的處理計算。變換方法的共同點：不會馬上觸發計算。每當調用一次變換方法，都會產生一個新的RDD。例如：map(func)，flatMap(func)ActionAction是RDD的算子，它的返回值不是一個RDD。Action操作是返回結果或者將結果寫入存儲的操作。Action是Spark應用啟動執行的觸發動作，得到RDD的相關計算結果或將RDD保存到文件系統中。CollectionRDDResultTransformationActionSparkConfSparkConf是用來對Spark進行任務參數配置的對象。是通過鍵值對的形式，設置Spark任務執行時所需要的參數。Spark讀取任務參數的優先級是：代碼配置>動態參數>配置文件。SparkContextSparkContext是Spark的入口，相當于應用程序的main函數。SparkContext表示與Spark集群的連接，可用于在該集群上創建RDD，記錄計算結果和環境配置等信息。DriverProgramSparkContextClusterManagerWorkerNodeExecutorCacheTaskTaskWorkerNodeExecutorCacheTaskTaskSparkSessionSpark2.0中引入了SparkSession的概念，為用戶提供了一個統一的切入點來使用Spark的各項功能。封裝了SparkConf和SparkContext對象，方便用戶使用Spark的各種API。SparkContextStreamingContextSQLContextSparkSessionSparkSQL簡介SparkSQL是Spark用來處理結構化數據的一個模塊，可以在Spark應用中直接使用SQL語句對數據進行操作。SQL語句通過SparkSQL模塊解析為RDD執行計劃，交給SparkCore執行。SparkCoreSparkSQLRDDSQL語句SparkSQL使用方式通過SparkSession提交SQL語句。任務像普通Spark應用一樣，提交到集群中分布式運行。JDBC：應用加載JDBC驅動，然后統一提交到集群的JDBCServer執行。JDBCServer是單點服務，會成為任務執行的瓶頸，不能處理海量數據和高并發任務。Spark應用ExecutorJDBCServerDriverApplicationDriverApplicationJDBC應用SparkSQL關鍵概念DataSetDataSet：DataSet是一個由特定域的對象組成的強類型集合，可通過功能或關系操作并行轉換其中的對象DataSet以Catalyst邏輯執行計劃表示，并且數據以編碼的二進制形式存儲，不需要反序列化就可以執行sort、filter、shuffle等操作。Dataset是“懶惰”的，只在執行action操作時觸發計算。當執行action操作時，Spark用查詢優化程序來優化邏輯計劃，并生成一個高效的并行分布式的物理計SparkSQL使用場景適合：結構化數據處理。對數據處理的實時性要求不高的場景需要處理PB級的大容量數據。不適合：實時數據查詢。SparkSQL簡單查詢查詢：帶條件的查詢：排序查詢：df.select("id","name").show()df.select($"id",$"name").where($"name"==="bbb").show()df.select($"id",$"name").orderBy($"name".desc).showdf.select($"id",$"name").sort($"name".desc).showSparkSQL開發場景說明：假定用戶有網民網購時停留網站的日志文本，基于某些業務要求，需要開發Spark應用程序并實現如下功能：統計日志文件中網購停留總時間超過2個小時的女性網民信息。日志文本log1.txt：網民停留日志日志文件第一列為姓名，第二列為性別，第三列為本次停留時間，單位為分鐘。分隔符為“,”。LiuYang,female,20YuanJing,male,10GuoYijun,male,5CaiXuyu,female,50Liyuan,male,20FangBo,female,50LiuYang,female,20YuanJing,male,10GuoYijun,male,50CaiXuyu,female,50FangBo,female,60LiuYang,female,20YuanJing,male,10CaiXuyu,female,50FangBo,female,50GuoYijun,male,5CaiXuyu,female,50Liyuan,male,20CaiXuyu,female,50FangBo,female,50開發思路目標統計日志文件中網購停留總時間超過2個小時的女性網民信息。大致步驟：創建表，將日志文件數據導入到表中。篩選女性網民，提取停留時間數據信息。匯總每個女性停留總時間。篩選出停留時間大于2個小時的女性網民信息。Scala樣例代碼objectCollectFemaleInfo{//表結構，后面用來將文本數據映射為dfcaseclassFemaleInfo(name:String,gender:String,stayTime:Int)defmain(args:Array[String]){//配置Spark應用名稱valsparkConf=newSparkConf().setAppName("FemaleInfo")valss=SparkSession.builder().config(sparkconf).getOrCreate()valsc=ss.sparkContextimportss.implicits._//通過隱式轉換，將RDD轉換成DataFrame，然后注冊表sc.textFile(args(0)).map(_.split(",")).map(p=>FemaleInfo(p(0),p(1),p(2).trim.toInt)).toDF.registerTempTable("FemaleInfoTable")//通過sql語句篩選女性上網時間數據,對相同名字行進行聚合valfemaleTimeInfo=sqlContext.sql(“

selectname,sum(stayTime)asstayTimefromFemaleInfoTablewheregender='female'groupbyname")//篩選出時間大于兩個小時的女性網民信息，并輸出valc=femaleTimeInfo.filter("stayTime>=120").collect()c.foreach(println)sc.stop()

}}Yarn-cluster作業提交打jar包上傳到Linux某個目錄下Yarn-cluster提交方式./spark-submit--masteryarn-cluster--classcom.huawei.bigdata.spark.CollectFemaleInfo../lib/spark-examples.jar/user/log1.txtSpark的應用場景有哪些?()查詢分析流式處理迭代計算數據挖掘Spark2.0中引入了SparkSession的概念，為用戶提供了一個統一的切入點來使用Spark的各項功能。()SparkSQL可以進行實時查詢。()

離線處理方案離線處理技術框架介紹數據存儲HDFS數據倉庫Hive離線分析SparkSQL數據采集工具離線批處理實戰常用采集工具由于大數據的數據源各種各樣，由此對數據采集的挑戰變的尤為突出。這里介紹幾款常用數據采集工具：SqoopLoaderSqoop簡介Sqoop項目開始于2009年，最早是作為Hadoop的一個第三方模塊存在，后來為了讓使用者能夠快速部署，也為了讓開發人員能夠更快速的迭代開發，Sqoop獨立成為一個Apache項目。Sqoop是一款開源的工具，主要用于在Hadoop(Hive)與傳統的數據庫(MySQL、PostgreSQL...)間進行數據的傳遞，可以將一個關系型數據庫（例如：MySQL,Oracle,PostgreSQL等）中的數據導進到Hadoop的HDFS中，也可以將HDFS的數據導進到關系型數據庫中。Sqoop應用RDBMS(MySQL,Oracle,PostgreSQL,DB2)Hadoop(HDFS,Hive,HBase)ImportExportSqoop原理(1)SqoopImport原理：Sqoop在import時，需要指定split-by參數。Sqoop根據不同的split-by參數值來進行切分，然后將切分出來的區域分配到不同map中。每個map中再處理數據庫中獲取的一行一行的值，寫入到HDFS中。同時split-by根據不同的參數類型有不同的切分方法，如比較簡單的int型，Sqoop會取最大和最小split-by字段值，然后根據傳入的num-mappers來確定劃分幾個區域。Sqoop原理(2)Sqoopexport原理：獲取導出表的schema、meta信息，和Hadoop中的字段match；并行導入數據：將Had