Spark的應用現(xiàn)狀

Spark需求背景

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

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

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

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

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

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

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

以Spark為核心的數(shù)據(jù)平臺結構

 

 

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

Spark的幾種典型應用

基于SparkStreaming的實時處理需求

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

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

 

 

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

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

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

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

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

 

 

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

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

由于之前大部分數(shù)據(jù)分析工作都是通過使用hive命令行完成的，為了將遷移至SparkSQL的代價最小，360系統(tǒng)部的同事開發(fā)了SparkSQL的命令行版本spark-hive。原有的以hive 命令運行的腳本，簡單的改成spark-hive便可以運行。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ù)的結構信息，即schema。同時，與Hive類似，DataFrame也支持嵌套數(shù)據(jù)類型(struct、array和map)。從API易用性的角度上 看，DataFrame API提供的是一套高層的關系操作，比函數(shù)式的RDD API要更加友好，門檻更低。

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

 

 

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

 

 

基于MLLib的機器學習需求

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

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

 

 

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

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

• 模型訓練起來速度比較快，時間可控。

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

 

 

部分經(jīng)驗總結

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

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

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

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

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

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

 

 

 

 

 

 

SparkSQL中使用Parquet

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

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

• SparSQL的Parquet的幾個操作：

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

2)讀取Parquet格式的文件

 

 

3)保存為Parquet格式文件

 

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

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

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

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

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

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