1、流式數(shù)據(jù)處理在數(shù)據(jù)工廠的應(yīng)用技術(shù)創(chuàng)新，變革未來CONTENT 目錄流式數(shù)據(jù)處理在百度的實踐百度數(shù)據(jù)工廠介紹流式數(shù)據(jù)處理在百度數(shù)據(jù)工廠的應(yīng)用CD總結(jié)PART A百 度 數(shù) 據(jù) 工 廠 介 紹百度數(shù)據(jù)工廠的發(fā)展史AHive離線批量數(shù)據(jù)分析PB級別查詢核心數(shù)據(jù)報表產(chǎn)出Spark1.6數(shù)據(jù)工廠數(shù)據(jù)挖掘分析數(shù)據(jù)實時分析數(shù)據(jù)復(fù)雜處理統(tǒng)一元數(shù)據(jù)管理統(tǒng)一計算資源調(diào)度統(tǒng)一計算引擎支持多種作業(yè)提交方式安全管理百度數(shù)據(jù)工廠整體介紹A集成數(shù)據(jù)加工處理環(huán)境工作空間工作流調(diào)度流式作業(yè)計算大數(shù)據(jù)計算引擎統(tǒng)一存儲訪問（統(tǒng)一元數(shù)據(jù)管理）統(tǒng)一計算資源調(diào)度引擎HDFS/BOSJDBCHBase/KafkaHiveYarnK8SSt

2、andaloneVM百度內(nèi)部日例行任務(wù)數(shù)：10W+日傳輸量：8PB+應(yīng)用業(yè)務(wù)數(shù)：200+ToB瑞聲、大地等多家大型企業(yè)PART B流 式 數(shù) 據(jù) 處 理 在 百 度 數(shù) 據(jù) 工 廠 應(yīng) 用S pa r k 流式S Q L 問題 實時轉(zhuǎn)離線問題實時轉(zhuǎn)大屏展示問題Spark流式SQL問題PART BS p a r k 未提供流式S Q LBspark.read.format(“json”).option(.).load().select().write.format(“csv”).save(outputPath)spark.readStream.format(“kafka”).load().sel

3、ect().writeStream.format(“csv”).start()批量Query流式Queryinsert into csv_table select *from json_table?APISQL從流式A P I 到流式S Q L 語法的定義B數(shù)據(jù)源映射讀取數(shù)據(jù)源 - 讀取表寫入數(shù)據(jù)源 - 寫入表數(shù)據(jù)處理映射數(shù)據(jù)處理 - SELECT/JOIN/UNION等增加Stream關(guān)鍵字區(qū)分批量SQL和流式SQLinsert into csv_table select stream * from kafa_tablespark.readStream.format(“kafka”).loa

4、d().select().writeStream.format(“csv”).start()流式S Q L 的統(tǒng)一元數(shù)據(jù)存儲Bcreate table kafka_table (name string) using kafka options()MetaTable Schema Parameters StorageProperties.以Properties存儲數(shù)據(jù)源配置保證表的通用性一張表對應(yīng)于一個數(shù)據(jù)源多個數(shù)據(jù)源會影響通用性只能定義通用配置例如：watermark配置并不是通用配 置，不能定義在表內(nèi)允許流批2種方式讀取當(dāng)帶stream關(guān)鍵字時，表被翻譯為流 式讀?。环粗畡t翻譯為批量讀取流式

5、S Q L 語義層解析Binsert into hive_table select stream *from kafa_tableSQLStreamingSink+- Project *+- StreamingRelation()1. 更新語義解析規(guī)則FindDataSource完成流式表解析UnresolvedCatalogRelation - StreamingRelationInsertIntoTable - SQLStreamingSink2. 增加可執(zhí)行類SQLStreamingSink負(fù)責(zé)觸發(fā)流式任務(wù)執(zhí)行解析流式配置S t r e a mJ o i nB a t c h 語義解析問

6、題Binsert into csv_table select stream * from kafka_table t1 join mysql_table t2 on t1.id = t2.p_id+- Project *+- Join (t1.id,t2.p_id)+- Alias t1+- StreamingRelation+- Alias t2+- StreamingRelation典型場景：實時統(tǒng)計工廠工人生產(chǎn)速度，對Kafka(實時生產(chǎn)數(shù)據(jù))和MySQL(員工數(shù)據(jù))做join分析，從而獲得每段時 間內(nèi)每個工人的產(chǎn)量。按照目前的設(shè)計，在語義解析過程中，mysql_table會解析成Str

7、eamingRelation，無法完成正 常的Streaming Join Batch。SQLStreamingSink流式S Q L 的設(shè)計升級B新增流式表類型只有流式表才能解析成StreamingRelation 流式表保留了流批查詢的語義創(chuàng)建流式表在Option中添加Streaming標(biāo)識create table kafka_table(name string) using kafkaoptions( isStreaming=true,)可選方案1. 根據(jù)Source類型區(qū)分例如：Kafka是流式Source， MySQL是批量Sourcepass原因：大部分Source即可以當(dāng)流式也可

8、以當(dāng)批量2. 建表時指定起止offset例如：MySQL指定讀取截止的數(shù)量等pass原因：這和我們對表的定義相違背，我們要做的是 通用表，用戶創(chuàng)建表后大家都可以用。流式表的使用B1. 批量查詢流式表例如：select * from kafka_stream_table在此場景中，流式表會被當(dāng)做普通表對待在語義解析層kafka_stream_table將被解析為 KafkaRelation流式表LocalRelation批量查詢流式查詢流式表例如：select stream * from kafka_stream_table流式表在此場景中，流式表將轉(zhuǎn)換為StreamingRelationStr

9、eamingRelation流查詢升級后的S t r e a mJ o i nB a t c h 解析Binsert into csv_table select stream * from kafka_table t1 join mysql_table t2 on t1.id = t2.p_idSQLStreamingSink+- Join (t1.id,t2.p_id)+- Project *+- StreamingRelation+- Project*+- LocalRelation升級：kafka_table創(chuàng)建為流式表，mysql_table創(chuàng)建為普通表流批統(tǒng)一S Q L 問題的總結(jié)B

10、統(tǒng)一的表元數(shù)據(jù)存儲流式SQL解析升級FindDataSourceinsertIntoTable - SQLStreamingSinkfrom Table- StreamingRelationStreaming Join Batch處理使用Stream/Normal Table區(qū)分表的使用實時轉(zhuǎn)離線問題PART B目前實時轉(zhuǎn)離線存在的問題B輸出信息缺乏管理輸出列信息？列分隔符？已輸出分區(qū)有哪些？遷移升級代價較大集群遷移(從測試集群遷移到正式集群)時，需要修改代碼才 能正常運(yùn)行擴(kuò)展文件格式繁雜如果甲方要求輸出sequenceFile，并指定了輸出格式怎么辦？df.writeStream.forma

11、t(“csv”).partitionBy(“col1”, “col2”).option(path, “hdfs:/hdfsPath“).start()實時轉(zhuǎn)離線樣例Case百度數(shù)據(jù)工廠實時轉(zhuǎn)數(shù)倉方案B輸出信息管理規(guī)范依托Hive元數(shù)據(jù)管理輸出升級代價小一般情況下修改Hive即可完成升級擴(kuò)展文件格式簡單小批量寫Hive，特殊格式添加Jar包即可df.writeStream.format(hive).option(table, employee).option(database, default).start()df.writeStream.format(“csv”).partitionBy(“c

12、ol1”, “col2”).option(path, “hdfs:/hdfsPath“).start()實時轉(zhuǎn)數(shù)倉的原理介紹B具體的解決方案：以實時轉(zhuǎn)離線為基礎(chǔ)在解析生成FileSink時讀取輸出表，將配置信息、HiveDynamicPartitionSinker注入到FileSink 中在小批量數(shù)據(jù)產(chǎn)出后提取輸出目錄的分區(qū)信息，并將分區(qū)信息提 供給HiveDynamicPartitionSinker，由HiveDynamicPartitionSinker添加Partition實時轉(zhuǎn)大屏展示問題PART B實時流輸出大屏存在的問題BSpark集群Kafka集群OLAP/ JDBC大屏展示通用的

13、實時流轉(zhuǎn)OLAP/JDBC接大屏展示方案統(tǒng)計日志需額外部署其他組件輸出到OLAP，需要部署OLAP服務(wù)甚至需要kafka集群從計算產(chǎn)出到最終展示經(jīng)多個系統(tǒng)，系統(tǒng)間延遲影響大實時流寫入OLAP的延遲產(chǎn)出數(shù)據(jù)中間落磁盤一般OLAP是基于磁盤的，讀寫OLAP時會產(chǎn)生I/O操作百度數(shù)據(jù)工廠實時流直接對接大屏方案BSpark集群kafka集群大屏展示統(tǒng)計日志Spark MemorySink + Livy JDBC API方案數(shù)據(jù)產(chǎn)出通過MemorySink 直接寫入Spark集群內(nèi)存通過Livy啟動交互式分析任務(wù)數(shù)據(jù)產(chǎn)出到內(nèi)存，無需落盤大屏通過JDBC接口直接獲取內(nèi)存數(shù)據(jù)前端通過JDBC API直接查詢

14、Spark集群內(nèi)存的數(shù)據(jù)百度數(shù)據(jù)工廠的統(tǒng)一計算引擎B用戶數(shù)據(jù)流-流流-批批-批元數(shù)據(jù) 掛載消息隊列文件上傳百度數(shù)據(jù)工廠Hive線上KafkaJDBCPART C流 式 數(shù) 據(jù) 處 理 在 百 度 數(shù) 據(jù) 工 廠 的 實 踐流式產(chǎn)品化頁面 流式數(shù)據(jù)處理在廣告物料分析的實踐百度數(shù)據(jù)工廠的統(tǒng)一計算引擎CStreaming SQL提交頁面Streaming實時監(jiān)控頁面廣告物料分析實踐案例C案例：在實際的產(chǎn)品運(yùn)維過程中，存在廣告主投放廣告的場景。廣告主投放廣告付錢是要看真實的點(diǎn)擊率、曝光率和轉(zhuǎn)化率的，而且很多廣告主是根據(jù) 曝光量、點(diǎn)擊量、轉(zhuǎn)化量來付錢的。這種情況下，我們就需要專門針對廣告物料進(jìn)行分析，根

15、據(jù)點(diǎn)擊、曝光 日志和轉(zhuǎn)化數(shù)據(jù)生成廣告的pv、uv、點(diǎn)擊率和轉(zhuǎn)化率，并根據(jù)計費(fèi)數(shù)據(jù)生成 廣告收益。廣告物料分析實踐案例C原始打點(diǎn)日志轉(zhuǎn)化表單數(shù)據(jù)KafkaKafkaHiveOLAPHiveKafka展示SQLStreamingStreamingHiveSink批量計算實時流轉(zhuǎn)數(shù)倉實時流直接大屏展示離線數(shù)據(jù)處理統(tǒng)計天級別的pv、uv等計費(fèi)數(shù)據(jù)產(chǎn)出廣告的pv、uv 點(diǎn)擊率、轉(zhuǎn)化率等圖表對曝光、點(diǎn)擊類打點(diǎn)數(shù)據(jù)和轉(zhuǎn)化 表單數(shù)據(jù)進(jìn)行分析整理，并整合計費(fèi) 數(shù)據(jù)產(chǎn)出實際廣告收益廣告物料分析實踐案例成果C實時流轉(zhuǎn)數(shù)倉延遲圖廣告物料報表實時展示上: 每個投放渠道的pv統(tǒng)計 左下: 信息流渠道的消費(fèi)情況 右下: 信息流渠道的消費(fèi)占比PART D總 結(jié)總