項目五

電商產品數據實時分析2023年2月創建數據流目錄Content1使用SparkStreaming對電商產品數據實時分析2離線數據分析處理速度慢項目導言零實時數據分析學習目標零知識目標了解流式計算的特征及優點；熟悉SparkStreaming架構；掌握DStream的創建操作；掌握DStream轉化操作。技能目標具備使用不同數據源創建DStream的能力；具備使用對數據進行轉換操作的能力；具備實現數據實時分析的能力。素養目標具有考慮問題時的換位思考能力；具有解決問題時的逆向思維能力；具有較強的總結能力。任務5-1：使用Hive創建電商產品數據庫壹流式計算簡介任務技能流計算在大數據的場景中非常常見，流計算由兩個部分組成分別是流數據和流計算DStream簡介DStream表示一個連續不間斷的數據流，DStream是隨時間推移而收到的數據序列SparkStreaming簡介SparkStreaming是Spark中的分布式流處理框架，能夠通過指定的時間間隔對數據進行處理，其最小時間間隔可達到500ms，SparkStreaming具有高吞吐量和容錯能力強DStream創建DStream的創建是在整個SparkStreaming程序中的，因此我們在創建DStream之前需要創建SparkStreaming對象，通過SparkStreaming對象創建DStream任務5-1：使用Hive創建電商產品數據庫壹1流式計算簡介流數據是指在時間分布和數量上無限的一系列動態數據的集合，這種數據的價值會隨著時間的推移而降低，所以就需要對其進行實時的數據分析并且做出毫秒級別的快速響應，否則就會失去數據原本存在意義數據流動速度快且持續，潛在大小也許是無窮無盡的；數據源較多，且數據格式復雜；數據量大，一旦經過處理，要么被丟棄，要么被歸檔存儲于數據倉庫；注重數據的整體價值，不過分關注個別數據；數據順序顛倒，或者不完整，系統無法控制將要處理的新到達的數據元素的順序。（1）流數據具有的特征任務5-1：使用Hive創建電商產品數據庫壹IBMInfoSphereStreams是IBM公司開發的業內先進流式計算軟件，支持開發和執行對數據流中的信息進行處理的應用程序。InfoSphereStreams

支持連續且快速地分析流數據（2）商業級流式計算框架任務5-1：使用Hive創建電商產品數據庫壹Storm是一款有Twitter開源的分布式實時大數據處理框架，用于實時分析數據，持續計算（3）開源流計算框架SparkStreaming是Spark體系中的一個流式處理框架，可以實現高吞吐量的、具備容錯機制的實時流數據的處理任務5-1：使用Hive創建電商產品數據庫壹FacebookPuma：FaceBook公司的實時數據處理分析框架，使用puma和HBase相結合來處理實時數據（4）公司為支持自身業務開發的流計算框架任務5-1：使用Hive創建電商產品數據庫壹2SparkStreaming簡介SparkStreaming是Spark中的分布式流處理框架，能夠通過指定的時間間隔對數據進行處理，其最小時間間隔可達到500ms，SparkStreaming具有高吞吐量和容錯能力強，支持多種數據源如Kafka、Flume、Twitter、ZeroMQ等（1）SparkStreaming處理的數據流圖任務5-1：使用Hive創建電商產品數據庫壹（2）SparkStreaming數據處理流程SparkStreaming是在對接外部數據流后按照時間間隔將數據劃分為batch(小批次數據流)供后續Sparkengine處理，所以實際上,SparkStreaming是按一個個batch(小批次)來處理數據流的任務5-1：使用Hive創建電商產品數據庫壹（3）SparkStreaming應用場景在SparkStreaming中，有無狀態操作、有狀態操作和窗口操作三種應用場景1、狀態操作針對于當前時間間隔內新生成的小批次數據，所有計算都只是基于這個批次的數據進行處理2、有狀態操作有狀態操作是指除需要當前生成的小批次數據外，還需要使用所有的歷史數據，即相當于統計總銷售量或銷售額等指標3、窗口操作SparkStreaming支持窗口計算以及在一個滑動窗口上進行數據的轉換操作任務5-1：使用Hive創建電商產品數據庫壹3DStream簡介DStream是隨時間推移而收到的數據序列。在內部，每個時間區間收到的數據都作為RDD存在，而DStream就是由這些RDD所組成的序列。SDstream數據可通過外部輸入源獲取（1）DStream內部實現任務5-1：使用Hive創建電商產品數據庫壹（2）DStream數據操作流程對DStream中數據的相關操作實際上就是對DStream內部的RDD進行的，通過設置時間，這個操作每隔一段時間就會對RDD進行操作并生成作為新的DStream中該時間段的RDD，在經過一系列操作后，可以將計算結果存儲到外部文件系統中，包括本地文件、HDFS、數據庫等任務5-1：使用Hive創建電商產品數據庫壹DStream的創建是在整個SparkStreaming程序中的，因此在創建DStream之前需要創建SparkStreaming對象，通過SparkStreaming對象創建DStream4DStream創建frompysparkimportSparkConfsc=SparkContext(master,appName)（1）SparkContext創建參數描述masterSpark、Mesos或YARN集群URL，或者是在本地模式下運行的特殊“local[*]”字符串appName應用程序在集群UI上顯示的名稱任務5-1：使用Hive創建電商產品數據庫壹（2）創建SparkStreaming對象frompyspark.streamingimportStreamingContextssc=StreamingContext(sc,Seconds)參數描述scSparkConf實例Seconds處理數據的時間間隔，單位為秒。StreamingContext對象創建完成后即可使用該對象中提供的不同數據源獲取方法創建DStream任務5-1：使用Hive創建電商產品數據庫壹使用Python編寫Spark程序Step1導入SparkContext與StreamingContext兩個包Step2創建StreamingContext對象，設置每間隔5秒讀取一次數據Step3啟動SparkStreamin程序，開始對目錄進行監控并打印數據Step4創建DStream使用SparkStreaming監控電商產品數據的處理與分析項目中的評價數據任務5-2：使用Hve對商產品數據統計貳DStream轉換操作任務技能DStream的轉換操作主要用于對所包含的數據進行處理和統計，包括過濾、合并、計算元素數量、集合、出現頻次DStream輸出操作在SparkStreaming中，DStream的輸出操作用于觸發DStream的轉換操作和窗口操作，是SparkStreaming程序必不可少的DStream窗口操作SparkStreaming還提供了窗口計算，能夠在數據的滑動窗口上應用轉換操作SparkStreaming啟動與停止SparkStreaming程序在編寫完成后，并不會被執行，啟動程序后，計算結束完畢程序也不會停止，需要人為設定任務5-2：使用Hve對商產品數據統計貳1算術運算DStream的轉換操作主要用于對所包含的數據進行處理和統計，包括過濾、合并、計算元素數量、集合、出現頻次方法描述map(func)對DStream中包含的每一個元素應用這個指定的函數，并以DStream格式返回結果flatMap(func)與map方法類似，只不過各個輸入項可以被輸出為零個或多個輸出項filter(func)對DStream的每一個數據應用條件函數進行判斷，當符合條件則加入新的DStream中，不符合的則刪除reduceByKey(func)與reduce()方法功能相同，但reduceByKey()方法針對（k,v）形式元素進行統計reduce(func)通過指定函數對DStream中的每一個元素進行聚合操作，然后返回只有一個元素的RDD構成的新的DStreamtransform(func)通過指定函數對DStream中的每一個元素執行指定操作，可以是任意的RDD操作，從而返回一個新的RDDcountByValue()計算DStream中每個RDD內的元素出現的頻次并返回新的DStream[(K,Long)]，其中K是RDD中元素的類型，Long是元素出現的頻次count()對DStream中包含的元素數量進行計數，返回一個內部只包含一個元素的RDD的DStreaamunion(otherStream)連接兩個DStream中的數據生成一個新的DStream任務5-2：使用Hve對商產品數據統計貳2DStream窗口操作SparkStreaming還提供了窗口計算，能夠在數據的滑動窗口上應用轉換操作方法描述window()該方法接收兩個參數，第一個參數為窗口長度，單位為秒；第二個參數為滑動時間間隔，單位為秒，并且不管是窗口長度還是滑動時間間隔都必須為創建StreamingContext對象設置時間的倍數countByWindow()統計滑動窗口的DStream中元素的數量，并以DStream格式返回，接受參數及代表意義與window()方法相同reduceByWindow()對滑動窗口中DStream的元素進行聚合操作，以DStream格式返回操作結果，該方法需要傳入三個參數，第一個參數即為進行聚合操作的函數，第二、三個參數與以上兩種方法的參數相同countByValueAndWindow()統計當前滑動窗口中DStream元素出現的頻率，并以DStream[(K,Long)]格式返回，其中K是元素的類型，Long是元素出現的頻次，接收參數與reduceByWindow()相同reduceByKeyAndWindow()對滑動窗口中DStream的（k,v）類型元素進行聚合操作，該方法包含四個參數，第一個參數為指定的聚合函數；第二個參數同樣是一個函數，但其用來處理流出的RDD，可不使用；第三個參數為窗口長度，單位為秒；第四個參數為滑動時間間隔，單位為秒任務5-2：使用Hve對商產品數據統計貳3DStream輸出操作在SparkStreaming中，DStream的輸出操作用于觸發DStream的轉換操作和窗口操作，是SparkStreaming程序必不可少的，可以將DStream中的數據保存到外部系統中，包括MySQL數據庫、本地文件、HDFS等方法描述pprint()DStream中每批數據的前十個元素。saveAsTextFiles(prefix,[suffix])將DStream中的數據以文本的形式保存在本地文件或HDFS中，其接受兩個參數，第一個參數為文件的路徑及名稱前綴，第二個參數為文件的格式，并且每隔規定時間都會生成一個文件名稱包含時間戳的本地文件foreachRDD(func)DStream數據推送到外部系統，通常用于實現將DStream數據保存到數據庫中任務5-2：使用Hve對商產品數據統計貳4SparkStreaming啟動SparkStreaming程序在編寫完成后，并不會被執行，DStream的相關操作只創建執行流程，設定了執行計劃，需要SparkStreaming的運行操作才會啟動SparkStreaming程序執行預期操作。啟動程序后，計算結束完畢程序也不會停止，只能通過相關方法手動停止程序（1）SparkStreaming的啟動與停止方法方法