[[337096]]

【51CTO.com快譯】

為什么轉(zhuǎn)向Spark?

雖然我們都在談?wù)摯髷?shù)據(jù)，但通常在職場(chǎng)闖蕩一段時(shí)間后才遇到大數(shù)據(jù)。在我供職的Wix.com，有逾1.6億個(gè)用戶在生成大量數(shù)據(jù)，因此需要擴(kuò)展我們的數(shù)據(jù)流程。

雖然有其他選擇(比如Dask)，但我們決定選擇Spark，原因主要有兩個(gè)：(1)它是目前的最新技術(shù)，廣泛用于大數(shù)據(jù)。(2)我們擁有Spark所需的基礎(chǔ)架構(gòu)。

如何針對(duì)pandas人群用PySpark編寫代碼?

您可能很熟悉pandas，僅僅搞好語(yǔ)法可能開(kāi)了個(gè)好頭，但確保PySpark項(xiàng)目成功還需要具備更多的條件，您要了解Spark的工作原理。

讓Spark正常工作很難，但一旦可以正常工作，它效果很棒!

Spark簡(jiǎn)述

建議看看這篇文章，閱讀MapReduce方面的說(shuō)明以便更深入的了解：《如何使用Spark處理大數(shù)據(jù)?》(https://towardsdatascience.com/the-hitchhikers-guide-to-handle-big-data-using-spark-90b9be0fe89a)。

我們?cè)谶@里要了解的概念是橫向擴(kuò)展。

從縱向擴(kuò)展入手比較容易。如果我們有一個(gè)運(yùn)行良好的pandas代碼，但后來(lái)數(shù)據(jù)對(duì)于它來(lái)說(shuō)太大了，我們可能會(huì)轉(zhuǎn)移到一臺(tái)內(nèi)存更多、功能更強(qiáng)的機(jī)器上，希望它能應(yīng)付得了。這意味著我們?nèi)杂幸慌_(tái)機(jī)器同時(shí)在處理全部數(shù)據(jù)——這就是縱向擴(kuò)展。

如果我們改而決定使用MapReduce，并將數(shù)據(jù)分成多個(gè)塊，然后讓不同的機(jī)器來(lái)處理每個(gè)塊，這就是橫向擴(kuò)展。

五個(gè)Spark最佳實(shí)踐

這五個(gè)Spark最佳實(shí)踐幫助我將運(yùn)行時(shí)間縮短至十分之一，并擴(kuò)展項(xiàng)目。

1. 從小處入手——采樣數(shù)據(jù)

如果我們想讓大數(shù)據(jù)起作用，先要使用少量數(shù)據(jù)看到我們方向正確。在我的項(xiàng)目中，我采樣10%的數(shù)據(jù)，并確保管道正常工作，這讓我可以使用Spark UI中的SQL部分，并查看數(shù)字流經(jīng)整個(gè)流程，不必等待太長(zhǎng)的時(shí)間來(lái)運(yùn)行流程。

憑我的經(jīng)驗(yàn)，如果您用小樣本就能達(dá)到所需的運(yùn)行時(shí)間，通?？梢暂p松擴(kuò)展。

2. 了解基礎(chǔ)部分：任務(wù)、分區(qū)和核心

這可能是使用Spark時(shí)要理解的最重要的一點(diǎn)：

1個(gè)分區(qū)用于在1個(gè)核心上運(yùn)行的1個(gè)任務(wù)。

您要始終了解自己有多少分區(qū)——密切關(guān)注每個(gè)階段的任務(wù)數(shù)量，并在Spark連接中將它們與正確數(shù)量的核心進(jìn)行匹配。幾個(gè)技巧和經(jīng)驗(yàn)法則可以幫助您做到這一點(diǎn)(所有這些都需要根據(jù)您的情況進(jìn)行測(cè)試)：

• 任務(wù)與核心之間的比例應(yīng)該是每個(gè)核心約2至4個(gè)任務(wù)。
• 每個(gè)分區(qū)的大小應(yīng)約為200MB–400MB，這取決于每個(gè)worker的內(nèi)存，可根據(jù)需要來(lái)調(diào)整。

3. 調(diào)試Spark

Spark使用惰性求值，這意味著它在等到動(dòng)作被調(diào)用后才執(zhí)行計(jì)算指令圖。動(dòng)作示例包括show()和count()等。

這樣一來(lái)，很難知道我們代碼中的bug以及需要優(yōu)化的地方。我發(fā)現(xiàn)大有幫助的一個(gè)實(shí)踐是，使用df.cache()將代碼劃分為幾個(gè)部分，然后使用df.count()強(qiáng)制Spark在每個(gè)部分計(jì)算df。

現(xiàn)在使用Spark UI，您可以查看每個(gè)部分的計(jì)算，并找出問(wèn)題所在。值得一提的是，如果不使用我們?cè)?1)中提到的采樣就使用這種做法，可能會(huì)創(chuàng)建很長(zhǎng)的運(yùn)行時(shí)間，到時(shí)將很難調(diào)試。

4. 查找和解決偏度

讓我們從定義偏度開(kāi)始。正如我們提到，我們的數(shù)據(jù)分到多個(gè)分區(qū);轉(zhuǎn)換后，每個(gè)分區(qū)的大小可能隨之變化。這會(huì)導(dǎo)致分區(qū)之間的大小出現(xiàn)很大的差異，這意味著我們的數(shù)據(jù)存在偏度。

可以通過(guò)在Spark UI中查看階段方面的細(xì)節(jié)，并尋找最大數(shù)和中位數(shù)之間的顯著差異以找到偏度：

圖1. 很大的差異(中位數(shù)= 3秒，最大數(shù)= 7.5分鐘)意味著數(shù)據(jù)有偏度。

這意味著我們有幾個(gè)任務(wù)比其他任務(wù)要慢得多。

為什么這不好——這可能導(dǎo)致其他階段等待這幾項(xiàng)任務(wù)，使核心處于等待狀態(tài)而無(wú)所事事。

如果您知道偏度來(lái)自何處，可以直接解決它并更改分區(qū)。如果您不知道/或沒(méi)辦法直接解決，嘗試以下操作：

調(diào)整任務(wù)與核心之間的比例

如前所述，如果擁有的任務(wù)比核心更多，我們希望當(dāng)更長(zhǎng)的任務(wù)運(yùn)行時(shí)，其他核心仍然忙于處理其他任務(wù)。盡管這是事實(shí)，但前面提到的比例(2-4：1)無(wú)法真正解決任務(wù)持續(xù)時(shí)間之間這么大的差異。我們可以試著將比例提高到10：1，看看是否有幫助，但是這種方法可能有其他缺點(diǎn)。

為數(shù)據(jù)加入隨機(jī)字符串(salting)

Salting是指用隨機(jī)密鑰對(duì)數(shù)據(jù)重新分區(qū)，以便可以平衡新分區(qū)。這是PySpark的代碼示例(使用通常會(huì)導(dǎo)致偏度的groupby)：

圖2

5. Spark中迭代代碼方面的問(wèn)題

這是個(gè)棘手的問(wèn)題。如前所述，Spark使用惰性求值，因此運(yùn)行代碼時(shí)，它僅構(gòu)建計(jì)算圖(DAG)。但當(dāng)您有一個(gè)迭代過(guò)程時(shí)，該方法可能會(huì)很成問(wèn)題，因?yàn)镈AG重新打開(kāi)了先前的迭代，而且變得很大。這可能太大了，驅(qū)動(dòng)程序在內(nèi)存中裝不下。由于應(yīng)用程序卡住了，因此很難找到問(wèn)題所在，但是在Spark UI中好像沒(méi)有作業(yè)在長(zhǎng)時(shí)間運(yùn)行(確實(shí)如此)，直到驅(qū)動(dòng)程序最終崩潰才發(fā)現(xiàn)并非如此。

這是目前Spark的一個(gè)固有問(wèn)題，對(duì)我來(lái)說(shuō)有用的解決方法是每5-6次迭代使用df.checkpoint()/ df.localCheckpoint()(試驗(yàn)一番可找到適合您的數(shù)字)。這招管用的原因是，checkpoint()打破了譜系和DAG(不像cache())，保存了結(jié)果，并從新的檢查點(diǎn)開(kāi)始。缺點(diǎn)在于，如果發(fā)生了什么岔子，您就沒(méi)有整個(gè)DAG來(lái)重新創(chuàng)建df。

原文標(biāo)題：5 Apache Spark Best Practices For Data Science，作者：Zion Badash

【51CTO譯稿，合作站點(diǎn)轉(zhuǎn)載請(qǐng)注明原文譯者和出處為51CTO.com】