Spark的應(yīng)用現(xiàn)狀

Spark需求背景

隨著數(shù)據(jù)規(guī)模的持續(xù)增長，數(shù)據(jù)需求越來越多，原有的以MapReduce為代表的Hadoop平臺越來越顯示出其局限性。主要體現(xiàn)在以下兩點(diǎn)：

•  任務(wù)執(zhí)行時間比較長。特別是某些復(fù)雜的SQL任務(wù)，或者一些復(fù)雜的機(jī)器學(xué)習(xí)迭代。

•  不能很好的支持像機(jī)器學(xué)習(xí)、實(shí)時處理這種新的大數(shù)據(jù)處理需求。

Spark作為新一代大數(shù)據(jù)處理的計(jì)算平臺，使得我們可以用Spark這一種平臺統(tǒng)一處理數(shù)據(jù)處理的各種復(fù)雜需求，非常好的支持了我們目前現(xiàn)有的業(yè)務(wù)。與原有MapReduce模型相比，其具有下面3個特點(diǎn)：

•  充分使用內(nèi)存作為框架計(jì)算過程存儲的介質(zhì)，與磁盤相比大大提高了數(shù)據(jù)讀取速度。利用內(nèi)存緩存，顯著降低算法迭代時頻繁讀取數(shù)據(jù)的開銷。

•  更好的DAG框架。原有在MapReduce M-R-M-R的模型，在Spark框架下，更類似與M-R-R,優(yōu)化掉無用流程節(jié)點(diǎn)。

•  豐富的組件支持。如支持對結(jié)構(gòu)化數(shù)據(jù)執(zhí)行SQL操作的組件Spark-SQL，支持實(shí)時處理的組件Spark-Streaming，支持機(jī)器學(xué)習(xí)的組件Mllib，支持圖形學(xué)習(xí)的Graphx。

以Spark為核心的數(shù)據(jù)平臺結(jié)構(gòu)

 

 

商業(yè)數(shù)據(jù)部的數(shù)據(jù)平臺架構(gòu)如上圖所示，Spark在其中起到一個非常核心作用。目前每天提交的Spark作業(yè)有1200多個，使用的資源數(shù)Max Resources: ,每日處理的數(shù)據(jù)量約有100TB。

Spark的幾種典型應(yīng)用

基于SparkStreaming的實(shí)時處理需求

商業(yè)數(shù)據(jù)部內(nèi)部有大量的實(shí)時數(shù)據(jù)處理需求，如實(shí)時廣告收入計(jì)算，實(shí)時線上ctr預(yù)估，實(shí)時廣告重定向等，目前主要通過SparkStreaming完成。

實(shí)時數(shù)據(jù)處理的第一步，需要有實(shí)時的數(shù)據(jù)。360的用戶產(chǎn)品，幾乎全國各地都部署有機(jī)房，主要有4大主力機(jī)房。實(shí)時數(shù)據(jù)的收集過程如下：

 

 

使用Apache flume實(shí)時將服務(wù)器的日志上傳至本地機(jī)房的Kafka，數(shù)據(jù)延遲在100ms以內(nèi)。

使用Kafka MirorMaker將各大主力機(jī)房的數(shù)據(jù)匯總至中心機(jī)房洛陽，數(shù)據(jù)延遲在200ms以內(nèi)。由于公司的網(wǎng)絡(luò)環(huán)境不是很好，為了保證低延遲，在MirorMaker機(jī)房的機(jī)器上，申請了帶寬的QOS保 證，以降低延遲。

數(shù)據(jù)處理的實(shí)時鏈路如下所示：

• 1種方式是通過Apache Flume實(shí)時寫入Hdfs，用于第二天全量數(shù)據(jù)的離線計(jì)算

• 1種方式是通過SparkSteaming實(shí)時處理，處理后數(shù)據(jù)會回流至Kafka或者Redis，便于后續(xù)流程使用。

 

 

基于SparkSQL和DataFrame的數(shù)據(jù)分析需求

SparkSQL是Spark的核心組件，作為新一代的SQL on Hadoop的解決方案，完美的支持了對現(xiàn)有Hive數(shù)據(jù)的存取。在與Hive進(jìn)行集成的同時，Spark SQL也提供了JDBC/ODBC接口，便于第三方工具如Tableau、Qlik等通過該接口接入Spark SQL。

由于之前大部分?jǐn)?shù)據(jù)分析工作都是通過使用hive命令行完成的，為了將遷移至SparkSQL的代價最小，360系統(tǒng)部的同事開發(fā)了SparkSQL的命令行版本spark-hive。原有的以hive 命令運(yùn)行的腳本，簡單的改成spark-hive便可以運(yùn)行。360系統(tǒng)部的同事也做了大量兼容性的工作。spark-hive目前已經(jīng)比較穩(wěn)定，成為數(shù)據(jù)分析的首選。

DataFrma是Spark 1.3引入的新API，與RDD類似，DataFrame也是一個分布式數(shù)據(jù)容器。

但與RDD不同的是，DataFrame除了數(shù)據(jù)以外，還掌握更多數(shù)據(jù)的結(jié)構(gòu)信息，即schema。同時，與Hive類似，DataFrame也支持嵌套數(shù)據(jù)類型(struct、array和map)。從API易用性的角度上 看，DataFrame API提供的是一套高層的關(guān)系操作，比函數(shù)式的RDD API要更加友好，門檻更低。

大數(shù)據(jù)開發(fā)過程中，可能會遇到各種類型的數(shù)據(jù)源，而DataFrame與生俱來就支持各種數(shù)據(jù)類型，如下圖，包括JSON文件、Parquet文件、Hive表格、本地文件系統(tǒng)、分布式文件系統(tǒng)(HDFS)以及云存儲(S3)。同時，配合JDBC，它還可以讀取外部關(guān)系型數(shù)據(jù)庫系統(tǒng)如Mysql，Oracle中的數(shù)據(jù)。對于自帶Schema的數(shù)據(jù)類型，如Parquet，DataFrame還能夠自動解析列類型。

 

 

通過組合使用DataFrame和SparkSQL，與MapReduce比較大大減少了代碼行數(shù)，同時執(zhí)行效率也得到了提升。如下示例是處理廣告主位置信息的scala代碼。

 

 

基于MLLib的機(jī)器學(xué)習(xí)需求

360DMP提供人群擴(kuò)展功能(Look-alike)。所謂人群擴(kuò)展，是基于廣告主創(chuàng)建的種子用戶，根據(jù)這些種子用戶的特征，挖掘、篩選、識別、拓展更多具有相似特征的用戶，以增加廣告的受眾。

業(yè)界的Look-alike有2種做法。第一種做法就是顯性的定位。廣告主先選中一部分種子用戶，根據(jù)種子用戶的標(biāo)簽再定位擴(kuò)展一部分其他用戶。比如如果種子用戶選擇的都是“化妝品-護(hù)膚”這個標(biāo)簽，那么根據(jù)這個標(biāo)簽可以找到其他的用戶，作為擴(kuò)展用戶。這種做法的缺點(diǎn)是不夠精確，擴(kuò)展出來的用戶過大。第二種方法是通過一個機(jī)器學(xué)習(xí)的模型，將問題轉(zhuǎn)化為機(jī)器學(xué)習(xí)模型，來定位廣告主的潛在用戶。我們采用的是這種方法。

 

 

在做Look-alike的過程中，用到了Spark中的Mlilib庫。MLlib算法庫的核心庫如上，選擇的是Classification中LR算法，主要原因有兩個：

• 模型比較簡單，易于理解和實(shí)現(xiàn)

• 模型訓(xùn)練起來速度比較快，時間可控。

LookAlike的第一步是建立模型。在這里，廣告主會首先提交一批種子用戶，作為機(jī)器學(xué)習(xí)的正樣本。其他的非種子用戶作為負(fù)樣本。于是問題就轉(zhuǎn)化為一個二分類的模型，正負(fù)樣本組成學(xué)習(xí)的樣本。訓(xùn)練模型之后，通過模型預(yù)測，最后得到廣告主需要的目標(biāo)人群。

 

 

部分經(jīng)驗(yàn)總結(jié)

使用Direct模式處理kafka數(shù)據(jù)

SparkStreaming讀取Kafka數(shù)據(jù)時，有兩種方法：Direct和Receiver。我們選擇的是Direct方法。與基于Receiver的方法相比，Direct具有以下優(yōu)點(diǎn)：

簡化并行性。無需創(chuàng)建多個輸入Kafka流和聯(lián)合它們。使用directStream，Spark Streaming將創(chuàng)建與要消費(fèi)的Kafka分區(qū)一樣多的RDD分區(qū)，這將從Kafka并行讀取數(shù)據(jù)。因此，Kafka和RDD分區(qū)之間存在一對一映射，這更容易理解和調(diào)整。

效率。在第一種方法中實(shí)現(xiàn)零數(shù)據(jù)丟失需要將數(shù)據(jù)存儲在預(yù)寫日志中，該日志進(jìn)一步復(fù)制數(shù)據(jù)。這實(shí)際上是低效的，因?yàn)閿?shù)據(jù)有效地被復(fù)制兩次。第二種方法消除了問題，因?yàn)闆]有接收器，因此不需要預(yù)寫日志。

Exactly-once語義。第一種方法使用Kafka的高級API在Zookeeper中存儲消耗的偏移量。這是傳統(tǒng)上消費(fèi)Kafka數(shù)據(jù)的方式。雖然這種方法(與預(yù)寫日志結(jié)合)可以確保零數(shù)據(jù)丟失(即至少一次語義)，但是一些記錄在一些故障下可能被消費(fèi)兩次，這是因?yàn)镾park Streaming可靠接收的數(shù)據(jù)與Zookeeper跟蹤的偏移之間存在不一致。因此，在第二種方法中，我們使用不基于Zookeeper的簡單的Kafka API，偏移由Spark Streaming在其檢查點(diǎn)內(nèi)跟蹤。這消除了Spark Streaming和Zookeeper / Kafka之間的不一致，所以每個記錄被Spark Streaming有效地接收一次。

Direct方法需要自己控制消費(fèi)的kafka offset，參考代碼如下。

 

 

 

 

 

 

SparkSQL中使用Parquet

相比傳統(tǒng)的行式存儲引擎，列式存儲引擎因其更高的壓縮比，更少的IO操作而越來越受到重視。這是因?yàn)樵诨ヂ?lián)網(wǎng)公司的大數(shù)據(jù)應(yīng)用中，大部分情況下，數(shù)據(jù)量很大并且數(shù)據(jù)字段數(shù)目比較多，但是大部分查詢只是查詢其中的部分行，部分列。這個時候，使用列式存儲就能極大的發(fā)揮其優(yōu)勢。

Parquet是Spark中優(yōu)先支持的列存方案。與使用文本相比，Parquet 讓 Spark SQL 的性能平均提高了 10 倍，這要感謝初級的讀取器過濾器、高效的執(zhí)行計(jì)劃，以及 Spark 1.6.0 中經(jīng)過改進(jìn)的掃描吞吐量。

• SparSQL的Parquet的幾個操作：

1)創(chuàng)建Parquet格式的Hive表

2)讀取Parquet格式的文件

 

 

3)保存為Parquet格式文件

 

Spark參數(shù)調(diào)優(yōu)

1)spark.sql.shuffle.partitions：在做Join或者Group的時候，可以通過適當(dāng)提高該值避免數(shù)據(jù)傾斜。

2)spark.testing.reserveMemory：Spark executor jvm啟動的時候，會默認(rèn)保留一部分內(nèi)存，默認(rèn)為300m。適當(dāng)?shù)臏p少這個值，可以增加 spark執(zhí)行時Storage Memory的值。設(shè)置方式是啟動spark shell的時候加上參數(shù)：--conf spark.testing.reservedMemory= 104857600。

3)spark.serializer:Spark內(nèi)部會涉及到很多對數(shù)據(jù)進(jìn)行序列化的地方，默認(rèn)使用的是Java的序列化機(jī)制。Spark同時支持使用Kryo序列化庫，Kryo序列化類庫的性能比Java序列化類庫的性能要高很多。官方介紹，Kryo序列化機(jī)制比Java序列化機(jī)制，性能高10倍左右。Spark之所以默認(rèn)沒有使用Kryo作為序列化類庫，是因?yàn)镵ryo要求最好要注冊所有需要進(jìn)行序列化的自定義類型，因此對于開發(fā)者來說，這種方式比較麻煩。設(shè)置方法是conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")。

標(biāo)簽： Mysql 大數(shù)據(jù) 大數(shù)據(jù)處理 大數(shù)據(jù)開發(fā) 大數(shù)據(jù)應(yīng)用 代碼 服務(wù)器 互聯(lián)網(wǎng) 互聯(lián)網(wǎng)公司 機(jī)房 腳本 開發(fā)者 數(shù)據(jù)分析 數(shù)據(jù)庫 網(wǎng)絡(luò)

版權(quán)申明：本站文章部分自網(wǎng)絡(luò)，如有侵權(quán)，請聯(lián)系：west999com@outlook.com
特別注意：本站所有轉(zhuǎn)載文章言論不代表本站觀點(diǎn)！
本站所提供的圖片等素材，版權(quán)歸原作者所有，如需使用，請與原作者聯(lián)系。