1、完整的大數(shù)據(jù)解決方案ABOUT MEAGENDASpark 介紹Spark 和 MongoDB案例演示唐建法/ TJMongoDB 高級方案架構(gòu)師MongoDB 中文社區(qū)聯(lián)合發(fā)起人Spark 介紹什么是Spark ?通用，快速，大規(guī)模數(shù)據(jù)處理引擎橫向擴(kuò)展，集團(tuán)作戰(zhàn)HDFS原生支持內(nèi)部循環(huán)數(shù)據(jù)流程全內(nèi)存計(jì)算可比Map Reduce 快100倍Spark SQL Streaming Machine LearningJava, Python, Scala, R我能用它做什么？流處理商業(yè)智能個(gè)性化產(chǎn)品推薦Spark 生態(tài)系統(tǒng)DataframePigSparkStand AloneHDFSSparkSt

2、and AloneYARNSparkStreamingMesosMesosHadoopMapReduceSparkHiveSparkSQL程序接口計(jì)算引擎資源管理數(shù)據(jù)存儲(chǔ)Spark 生態(tài)系統(tǒng)DataframePigSparkStand AloneHDFSSparkStand AloneYARNSparkStreamingMesosMesosHadoopMapReduceSparkHiveSparkSQL程序接口計(jì)算引擎資源管理數(shù)據(jù)存儲(chǔ)HDFS vs. MongoDBHDFSMongoDB橫向擴(kuò)展，支持TB-PB級數(shù)據(jù)量共同點(diǎn)低成本， x86數(shù)據(jù)自動(dòng)多份復(fù)制支持非結(jié)構(gòu)化數(shù)據(jù)粗顆粒度存儲(chǔ)細(xì)顆粒度，

3、結(jié)構(gòu)化存儲(chǔ)差異點(diǎn)無索引二級索引一次寫入，多次讀讀寫混合非交互式，離線交互式，實(shí)時(shí)在線分鐘級SLA毫秒級SLA一個(gè)日志的例子索引 error: 1 ts: 2016-07-31 23:50:50, host: xyz, ts: 2016-07-31 23:49:23, host: def, ts: 2016-07-31 23:49:22, host: xyz,04, .body: 19, .body: ll, body: 2016-07-31.log. . . ts: 2016-07-01 02:04:12, host: abc,0, body: 2016-07-30.log. . .2016-

4、07-01.log使用場景HDFSMongoDB7月1日到31日所有頁面的點(diǎn)擊量統(tǒng)計(jì)OKOK每日HTTP 404錯(cuò)誤日志數(shù)量統(tǒng)計(jì)低效：需要掃描所有文件行可利用索引秒級響應(yīng)對日志行增加自定義字段保存分析結(jié)果不支持OKerror:4 error:0 error:nuerror: 50Spark + MongoDBmapfilter union intersectSpark工作方式Result Result Result ResultParellelizeTransformActionParellelizeTransformActionParellelizeTransformActionParell

5、elizeTransformActionSpark MongoDB 架構(gòu)PROCESSINGDriver任務(wù)調(diào)度原始數(shù)據(jù)計(jì)算結(jié)果Spark Master連接器WORKERWORKERWORKERWORKEREXECUTOEXECUTEXECUTEXECUTRORORORSTORAGESpark MongoDB HDFS混合架構(gòu)PROCESSINGDriver任務(wù)調(diào)度原始數(shù)據(jù)計(jì)算結(jié)果Spark Master連接器WORKERWORKERWORKERWORKEREXECUTOEXECUTEXECUTEXECUTRORORORSTORAGEMongo Spark Connector 連接器雙向支持：

6、讀出與寫入條件下推Co-Lo 部署，本地?cái)?shù)據(jù)訪問/mongodb/mongo-spark案例分享Spark + MongoDB 成功案例客戶案例實(shí)現(xiàn)價(jià)值乘客行為大數(shù)據(jù)分析， 客戶360度視圖改善的客戶體驗(yàn)， 降低客戶流失實(shí)時(shí)監(jiān)控分析跨國銀行所有客戶交易行為降低客戶資金風(fēng)險(xiǎn)， 提高合規(guī)性運(yùn)價(jià)計(jì)算集群及緩存， 降低機(jī)票查詢響應(yīng)時(shí)間改善用戶使用體驗(yàn)， 提高直銷率東方航空的挑戰(zhàn)每天客座率每天需支持運(yùn)價(jià)查詢數(shù)量，基于12000：1查定比16 億130,000260,000目標(biāo)：50%直銷率NYCDateBOSDateLAXDateBOSLAXStay DurationD

7、ateNYCDate思路：空間換時(shí)間365 天x 1000 航班x 26 倉位x 100 渠道x N 類型=數(shù)十億運(yùn)價(jià)x 1000+LAXStay Duration實(shí)時(shí)運(yùn)算目前運(yùn)價(jià)實(shí)現(xiàn)為實(shí)時(shí)計(jì)算， 可以支持每天1000多萬QPSLAXStay DurationNYCStay DurationBOSStay Duration26 倉位PVGSZX窮舉所有日期航班倉位計(jì)算出所有運(yùn)價(jià)保存結(jié)果至MongoDB直接從MongoDB查詢運(yùn)價(jià)空間換時(shí)間Date365Spark + MongoDB 方案批處理計(jì)算 B2TSpark Master B2M運(yùn)價(jià)查詢Fare CacheSeat InventoryD

8、RVDRVDRVDRVB2CMobile.Fare APIFare Inv ImplInv APICall CenterSpark MasterIB2CFare APIFare Inv ImplInv APISpark Submit批處理計(jì)算流程準(zhǔn)備任務(wù)到MongoDBVars:Flight schedule Base price Price Rules讀出計(jì)算任務(wù)Spark 并發(fā)計(jì)算varsvars結(jié)果存入MongoDBInput jobInput joboutputoutputENDParallel ComputeCollect ResultsParallel ComputeCollect

9、 ResultsSplit JobsSTARTSubmit Batch JobMasterLoad Reference DataBroadcast VariablesSpark 任務(wù)入口程序/ initialization dependencies including base prices, pricing rules and some reference dataMap dependencies = MyDependencyManager.loadDependencies();/ broadcasting dependenciesjavaSparkContext.broadcast(dep

10、endencies);/ create job rddcabinsRDD = MongoSpark.load(javaSparkContext).withPipeline(pipeline)/ for each cabin, date, airport pair, calculate the pricecabinsRDD.map(function calc_price);/ collect the result, which will cause the data to be stored into MongoDBcabinsRDD.collect()cabinsRDD.saveToMongo

11、()處理能力和響應(yīng)時(shí)間比較單節(jié)點(diǎn)處理能力響應(yīng)時(shí)間3503500300300025025002002000LatencyThroughput150150010010005050000LegacySpark+MongoLegacySpark+MongoSpark MongoDB演示安裝Spark# #curl -OLmkdir -p tar -xvf/spark-1.6.0-bin-hadoop2.6.tgz/sparkspark-1.6.0-bin-hadoop2.6.tgz -C /spark -strip-componen

12、ts=1測試連接器# cd spark# ./bin/spark-shell -conf spark.mongodb.input.uri=mongodb://flights.av -conf spark.mongodb.output.uri=mongodb://flights.output -packages org.mongodb.spark:mongo-spark-connector_2.10:1.0.0import com.mongodb.spark._import org.bson.DocumentMongoSpark.load(sc).take(1

13、0).foreach(println)簡單分組統(tǒng)計(jì)數(shù)據(jù)： 365天，所有航班庫存信息，500萬文檔任務(wù)： 按航班統(tǒng)計(jì)一年內(nèi)所有余票量MongoSpark.load(sc).map(doc=(doc.getString(flight) ,doc.getLong(seats).reduceByKey(x,y)=(x+y).take(10).foreach(println)簡單分組統(tǒng)計(jì)加條件過濾數(shù)據(jù)： 365天，所有航班庫存信息，500萬文檔任務(wù)： 按航班統(tǒng)計(jì)一年內(nèi)所有庫存，但是只處理昆明出發(fā)的航班import org.bson.DocumentMongoSpark.load(sc).withPipeline(Seq(Document.parse( $match: orig :KMG ).map(doc=(doc.getString(flight) ,doc.getLong(seats).reduceByKey(x,y)=(x+y).take(10).foreach(println)結(jié)語性能優(yōu)化事項(xiàng)使用合適的chunksize (MB)Total data size / chunksize = chunks = RDD partitions =