Spark平臺(tái)在電信運(yùn)營商的應(yīng)用實(shí)踐.pdf

spark平臺(tái)在電信運(yùn)營商的應(yīng)用實(shí)踐 亞信大數(shù)據(jù)平臺(tái) 田毅 目錄 項(xiàng)目實(shí)踐分享 基于spark改造用戶標(biāo)簽分析查詢平臺(tái) 基于spark streaming改造內(nèi)容識別處理平臺(tái) 一些心得分享 如何用好external datasource api 高效的在spark streaming中引用外部數(shù)據(jù) 基于spark改造用戶標(biāo)簽分析查詢平臺(tái) 3 tcl腳本 基于spark改造用戶標(biāo)簽分析查詢平臺(tái) 改造前的設(shè)計(jì) 4 數(shù)據(jù)庫 指標(biāo)表 通信數(shù)據(jù) 上網(wǎng)數(shù)據(jù) 數(shù)據(jù)清洗指標(biāo)計(jì)算標(biāo)簽計(jì)算 標(biāo)簽表接口表 用戶 數(shù)據(jù)探索客戶群計(jì)算 sql 基于spark改造用戶標(biāo)簽分析查詢平臺(tái) 改造前的問題 1 標(biāo)簽數(shù)量越來越大，數(shù)據(jù)庫負(fù)載過高，擴(kuò)展成本高 2 標(biāo)簽表的列數(shù)隨著標(biāo)簽數(shù)量增加不斷增多，部分現(xiàn)場達(dá)到2000+，只能通過分表的方式解決，查 詢時(shí)需要join操作 3 標(biāo)簽與指標(biāo)的計(jì)算無法擺脫sql的約束，無法快速集成機(jī)器學(xué)習(xí)的算法 tcl腳本 基于spark改造用戶標(biāo)簽分析查詢平臺(tái) 第一次改造設(shè)計(jì): 小試牛刀 6 數(shù)據(jù)庫 指標(biāo)表 通信數(shù)據(jù) 上網(wǎng)數(shù)據(jù) 數(shù)據(jù)清洗指標(biāo)計(jì)算標(biāo)簽計(jì)算 標(biāo)簽表接口表 用戶 數(shù)據(jù)探索客戶群計(jì)算 sparksql hdfs 基于spark改造用戶標(biāo)簽分析查詢平臺(tái) 改造后的好處 1 使用sparksql parquet的方案，有效保證了查詢效率 2 原有系統(tǒng)基本不用太大改造 3 查詢系統(tǒng)具備平行擴(kuò)展能力 未解決的問題 1 標(biāo)簽與指標(biāo)的計(jì)算無法擺脫sql的約束，無法快速集成機(jī)器學(xué)習(xí)的算法 產(chǎn)生出來的新問題 1 增加了從數(shù)據(jù)庫倒出數(shù)據(jù)，加載到hdfs的額外步驟 2 增加了從文本數(shù)據(jù)轉(zhuǎn)化為parquet格式的額外步驟 sparksql 基于spark改造用戶標(biāo)簽分析查詢平臺(tái) 第二次改造設(shè)計(jì)：大刀闊斧 8 hdfs 指標(biāo)表 通信數(shù)據(jù) 上網(wǎng)數(shù)據(jù) 數(shù)據(jù)清洗指標(biāo)計(jì)算標(biāo)簽計(jì)算 標(biāo)簽表接口表 用戶 數(shù)據(jù)探索客戶群計(jì)算 sparksql 基于spark改造用戶標(biāo)簽分析查詢平臺(tái) 改造后的好處 1 通過sparksql替換掉了原有的數(shù)據(jù)庫，整個(gè)系統(tǒng)的擴(kuò)展性進(jìn)一步增強(qiáng) 2 兩套sparksql可以根據(jù)各自忙閑時(shí)的不同，共享整個(gè)系統(tǒng)的計(jì)算資源 遺留的問題 1 沒有擺脫標(biāo)簽分析算法對于sql的依賴 2 系統(tǒng)前端仍然依賴etl系統(tǒng)對數(shù)據(jù)進(jìn)行抽取加載 怎么破？ 基于spark改造用戶標(biāo)簽分析查詢平臺(tái) spark 1.3.0 發(fā)布了 external datasource api進(jìn)一步增強(qiáng) dataframe提供了豐富多樣的數(shù)據(jù)源支持 dataframe提供了一整套用于操縱數(shù)據(jù)的dsl 外部系統(tǒng) spark applications based on df 基于spark改造用戶標(biāo)簽分析查詢平臺(tái) 第三次改造設(shè)計(jì)：更進(jìn)一步 11 hdfs 指標(biāo)表 數(shù)據(jù)表1 數(shù)據(jù)表2 指標(biāo)計(jì)算 標(biāo)簽計(jì)算 標(biāo)簽表 用戶 數(shù)據(jù)探索客戶群計(jì)算 sparksql extdatasour ce 按需抽取數(shù) 據(jù) 原有sql轉(zhuǎn)化 為df的api 基于spark改造用戶標(biāo)簽分析查詢平臺(tái) 改造后的好處 1 終于擺脫了對sql的依賴，為后續(xù)引入復(fù)雜算法分析打下基礎(chǔ) 2 利用external datasource api可以根據(jù)計(jì)算需求從源表抽取指定的數(shù)據(jù) 3 基于df的處理程序代碼量僅有原程序的1/10，可讀性大大提高 遺留的問題 1 如何控制對源數(shù)據(jù)庫的壓力問題 = 時(shí)間窗 2 ext ds的實(shí)現(xiàn)對于不同的數(shù)據(jù)庫類型需要進(jìn)行細(xì)致的優(yōu)化 基于spark streaming改造內(nèi)容識別平臺(tái) 內(nèi)容識別平臺(tái)功能介紹 產(chǎn)品目標(biāo)：通過對上網(wǎng)日志的分析還原用戶上網(wǎng)時(shí)的場景 用戶上網(wǎng)記錄 入口識別： app還是瀏覽器 應(yīng)用識別： 微博，微信，uc url內(nèi)容識別： 新聞，體育 mapreduce job map task 基于spark streaming改造內(nèi)容識別平臺(tái) 改造前的設(shè)計(jì) 上網(wǎng)數(shù)據(jù)輸出目錄輸入目錄 distribute cache 數(shù)據(jù)分析 后續(xù)系統(tǒng) 標(biāo)簽分析 日志查詢 營銷活動(dòng) 識別規(guī)則 識別規(guī)則 基于spark streaming改造內(nèi)容識別平臺(tái) 改造前的問題 1 數(shù)據(jù)處理延遲較高 2 需要頻繁加載規(guī)則數(shù)據(jù)到內(nèi)存 3 數(shù)據(jù)源逐漸變?yōu)閷?shí)時(shí)接口 spark streaming dstream.map() 基于spark streaming改造內(nèi)容識別平臺(tái) 改造前的設(shè)計(jì) 上網(wǎng)數(shù)據(jù) kafka broadcast 數(shù)據(jù)分析 后續(xù)系統(tǒng) 標(biāo)簽分析 日志查詢 營銷活動(dòng) kafka input topic output topic 識別規(guī)則 將原有的hdfs 改為kafka接口 mr引擎換為 spark streaming 原有mr的map處理邏輯 遷移到dstream的map 方法 規(guī)則數(shù)據(jù)改為通過 廣播發(fā)布到所有的 executor 基于spark streaming改造內(nèi)容識別平臺(tái) 改造后的好處 1 數(shù)據(jù)分析的代碼邏輯幾乎沒有修改, 兼容了原有的hdfs文件接口 2 規(guī)則數(shù)據(jù)只需要一次加載，可以長期保存在executor的內(nèi)存中 3 通過kafka spark streaming實(shí)現(xiàn)了流式處理的要求 4 數(shù)據(jù)處理延遲從原有的分鐘級別降低到秒級 改造過程的經(jīng)驗(yàn) 1 序列化問題 = 使用kryo序列化需要注意先注冊 2 流處理框架和業(yè)務(wù)邏輯兩部分代碼建議完全隔離 3 業(yè)務(wù)邏輯可以保持java的實(shí)現(xiàn)方式，通過反射等方式調(diào)用業(yè)務(wù)邏輯的代碼 如何用好external datasource api external datasource api 是spark 1.2.0版本中一個(gè)重要的feature 賦予spark平臺(tái)高效靈活訪問外部數(shù)據(jù)源的能力 我們?nèi)粘５氖褂弥幸步?jīng)常會(huì)遇到數(shù)據(jù)存在于多個(gè)數(shù)據(jù)源之中的場景 如何使用ext ds的api來實(shí)現(xiàn)對多個(gè)數(shù)據(jù)源的支持呢？ 讓我們用hbase作為外部數(shù)據(jù)源舉個(gè)例子 18 如何用好external datasource api 網(wǎng)上已經(jīng)有很多hbase的ext ds的實(shí)現(xiàn)，如： /huawei-spark/hbase /apache/phoenix/tree/master/phoenix-spark 我們簡要的分析一下實(shí)現(xiàn)hbase的ext ds的幾個(gè)要點(diǎn) task task task task executor executor region region region region regionserver regionserver 如何才能達(dá)到這樣的實(shí)現(xiàn)呢？我們先來看看external ds的設(shè)計(jì)原理 如何用好external datasource api optimizer filter pushdown column pruning 解析為ext relation analyzer resolve relation strategy build physicalrdd 將filter盡可能push 到ext relation，同 時(shí)進(jìn)行column pruning sql spark plan 根據(jù)ext relation實(shí) 現(xiàn)，調(diào)用buildscan 生成rdd 如何用好external datasource api dagscheduler submit jobsubmit stagesubmit task external rdd job rdd.partitionsrdd. getpreferredlocations executor executor executor pute 高效的在spark streaming中引用外部數(shù)據(jù) 實(shí)現(xiàn)如何分partition 接口定義 rdd.scala /* * implemented by subclasses to return the set of partitions in this rdd. this method will only * be called once, so it is safe to implement a time-consuming computation in it. */ protected def getpartitions: arraypartition 實(shí)現(xiàn)思路： 根據(jù)createrelation方法中傳入的參數(shù)確定hbase的連接方式和表名 通過調(diào)用hbase的api獲取該表中所有region的列表 根據(jù)buildscan中傳入的arrayfilter對region進(jìn)行一定的過濾 根據(jù)region列表生成一個(gè) arraypartition 這樣做的好處： 每個(gè)partition只處理一個(gè)region的數(shù)據(jù)，為后面的getpreferredlocations做基礎(chǔ) 22 高效的在spark streaming中引用外部數(shù)據(jù) 實(shí)現(xiàn)getpreferredlocations方法 接口定義 rdd.scala /* * optionally overridden by subclasses to specify placement preferences. */ protected def getpreferredlocations(split: partition): seqstring = nil 實(shí)現(xiàn)思路： 根據(jù)split中包含的region信息，來確定這個(gè)region在哪個(gè)節(jié)點(diǎn) 這樣做的好處： task在調(diào)度的時(shí)候可以優(yōu)先被調(diào)度到region所在的機(jī)器上執(zhí)行，減少網(wǎng)絡(luò)傳輸 23 高效的在spark streaming中引用外部數(shù)據(jù) 實(shí)現(xiàn)compute方法 接口定義 rdd.scala /* * : developerapi : * implemented by subclasses to compute a given partition. */ developerapi def compute(split: partition, context: taskcontext): iteratort 實(shí)現(xiàn)思路： 根據(jù)split中包含的region信息、filter、requiredcolumns調(diào)用hbase的api進(jìn)行查詢 24 如何用好external datasource api 實(shí)現(xiàn)ext ds的幾個(gè)要點(diǎn) 1 繼承relationprovider實(shí)現(xiàn)一個(gè)createrelation方法，來獲取所有參數(shù) 2 繼承baserelation實(shí)現(xiàn)一個(gè)根據(jù)自定義的schema生成方法 3 根據(jù)數(shù)據(jù)源類型選擇實(shí)現(xiàn)不同級別的buildscan（還有insertablerelation可以實(shí)現(xiàn)插入） 4 在buildscan中根據(jù)push down的過濾條件生成rdd 5 rdd中實(shí)現(xiàn)如何分partition 6 rdd中實(shí)現(xiàn)getpreferredlocations方法 7 rdd中實(shí)現(xiàn)compute方法 高效的在spark streaming中引用外部數(shù)據(jù) 為什么要引用外部數(shù)據(jù) 流式處理中輸入數(shù)據(jù)包含的信息有限 大量的使用場景需要關(guān)聯(lián)外部數(shù)據(jù)進(jìn)行邏輯判斷處理 spark streaming自身提供的一些方法 只讀數(shù)據(jù)：broadcast 更新數(shù)據(jù)：updatestatebykey 存在的問題 外部數(shù)據(jù)過大無法broadcast updatestatebykey需要進(jìn)行shuffle updatestatebykey每次都輸出全量數(shù)據(jù) 26 高效的在spark streaming中引用外部數(shù)據(jù) 只讀數(shù)據(jù)大數(shù)據(jù) 特點(diǎn)：數(shù)據(jù)量巨大，無法通過broadcast方式發(fā)送到各個(gè)executor 場景：關(guān)聯(lián)外部維表數(shù)據(jù) 27 場景1: 如果一個(gè)節(jié)點(diǎn)上的內(nèi)存能放下 方案：本機(jī)內(nèi)存存儲(chǔ) executor work node executor executor executor redis 場景2: 如果一個(gè)節(jié)點(diǎn)上的內(nèi)存不能放下 方案：分布式內(nèi)存存儲(chǔ) executor work node executor executor executor proxy work node redis redis redis redis 高效的在spark streaming中引用外部數(shù)據(jù) 更新數(shù)據(jù) 特點(diǎn)：數(shù)據(jù)不分大小，需要按batch進(jìn)行更新 場景：較長時(shí)間周期內(nèi)的累計(jì)數(shù)據(jù)，如：連續(xù)30分鐘位置停留 28 場景1: 如果需要更新的數(shù)據(jù)較小 方案：cogroup saveasfile 場景2: 如果需要更新的數(shù)據(jù)較大 方案：分布式內(nèi)存存儲(chǔ) executor work node executor executor executor proxy work node redis redis redis redis hdfs rdd rdd dstream cogroup rdd rdd dstream 高效的在spark streaming中引用外部數(shù)據(jù) 使用分布式內(nèi)存存儲(chǔ)時(shí) 核心問題：如何減少訪問緩存的次數(shù) 29 方案1: 合并相同key的cache 方案2：mappartition中設(shè)立c