摘要：基本編程模式里有兩個很重要的概念一般簡稱為和，在上一篇文章中基本概念解析有講到。可以說，和貫穿了的大部分生命周期，從的初始化，到數(shù)據(jù)的清洗，計算，到最后獲取，展示結(jié)果。

本系列是綜合了自己在學(xué)習(xí)spark過程中的理解記錄 ＋ 對參考文章中的一些理解 ＋ 個人實踐spark過程中的一些心得而來。寫這樣一個系列僅僅是為了梳理個人學(xué)習(xí)spark的筆記記錄，所以一切以能夠理解為主，沒有必要的細(xì)節(jié)就不會記錄了，而且文中有時候會出現(xiàn)英文原版文檔，只要不影響理解，都不翻譯了。若想深入了解，最好閱讀參考文章和官方文檔。

其次，本系列是基于目前最新的 spark 1.6.0 系列開始的，spark 目前的更新速度很快，記錄一下版本好還是必要的。

最后，如果各位覺得內(nèi)容有誤，歡迎留言備注，所有留言 24 小時內(nèi)必定回復(fù)，非常感謝。

Tips: 如果插圖看起來不明顯，可以：1. 放大網(wǎng)頁；2. 新標(biāo)簽中打開圖片，查看原圖哦。

spark 里有兩個很重要的概念：SparkContext 一般簡稱為 sc] 和 RDD，在上一篇文章中 [『 Spark 』2. spark 基本概念解析 有講到?？梢哉f，sc 和 RDD 貫穿了 spark app 的大部分生命周期，從 app 的初始化，到數(shù)據(jù)的清洗，計算，到最后獲取，展示結(jié)果。

為了更加深入的了解 RDD 和基于 RDD 的編程模型，我們先把 RDD 的屬性簡單的分一個類，然后再通過一張流程圖來理解。

接觸過 RDD 的人肯定都知道 transform 和 action 這兩個核心概念，甚至很多人都認(rèn)為 RDD 僅僅有 transform 和 action 這兩個概念。殊不知其實 RDD 里面還有很多其他方法，下面我們來簡單的分個類，在看這里的時候最好參考一下官方的 api 文檔

RDD

action : count, take, sample, first, collect ...

transform : foreach, glom, map ...

method : cache, checkpoint, id, isCheckpointed, isEmpty, keys, lookup, max, mean, name, setName ...

property : context

看到了嗎，這里其實 RDD 其實有很多既不是 transform 也不是 action 的函數(shù)和屬性，在編寫 spark app 的時候，其實很多時候我們都會用到那些 method，這樣在開發(fā)調(diào)試過程中都會更加方便。比如說 cache, setName, lookup, id 這些，在開發(fā)過程中都很有用。

如圖所示，我們構(gòu)建 spark app，一般都是三個步驟:

加載數(shù)據(jù)集，這里的數(shù)據(jù)集大概分為兩組:

一種是不變的，靜態(tài)數(shù)據(jù)集，大多數(shù)場景都是從數(shù)據(jù)庫，文件系統(tǒng)上面加載進(jìn)來

另一種是動態(tài)的數(shù)據(jù)集，一般做 streaming 應(yīng)用的時候用到，大多數(shù)場景是通過 socket 來加載數(shù)據(jù)，復(fù)雜場景可以通過文件系統(tǒng)，akka actors，kafka，kinesis 和 一些第三方提供的 streaming api [twitter 等] 來作為數(shù)據(jù)源加載數(shù)據(jù)

處理數(shù)據(jù)，這是重點中的重點，不過不外乎都是從三個方面來完成這里的數(shù)據(jù)清理，邏輯運算等:

自定義的一些復(fù)雜處理函數(shù)或者第三方包 [下面我們稱為函數(shù)集]

通過 RDD 的 transform，action 和函數(shù)集來完成整個處理，計算流程

通過 RDD 提供的 cache，persist，checkpoint 方法把一些處理流程中的重要處理節(jié)點和常用數(shù)據(jù)緩存和備份，以加速處理，計算速度

結(jié)果展示，這里一般情況都是使用 RDD 的 collect，take，first，top 等方法把結(jié)果取出來，更常用的是先把結(jié)果取出來，放到一個數(shù)據(jù)庫或文件系統(tǒng)上，然后再提供給專門展示結(jié)果的另一個 application 使用。

下面我將從幾個方面來介紹這個例子：首先是介紹蒙特卡羅方法的基本概念和應(yīng)用，然后是介紹如何用蒙特卡羅方法來估算 pi 的值，最后是看在 spark 集群中如何用多種方法來實現(xiàn)一個蒙特卡洛應(yīng)用來計算 pi 的值。

from wiki:

Monte Carlo methods (or Monte Carlo experiments) are a broad class of computational algorithms that rely on repeated random sampling to obtain numerical results. They are often used in physical and mathematical problems and are most useful when it is difficult or impossible to use other mathematical methods. Monte Carlo methods are mainly used in three distinct problem classes:[1] optimization, numerical integration, and generating draws from a probability distribution.

總的來說，蒙特卡羅是一種基于隨機(jī)樣本實驗來進(jìn)行估值的一種計算方法。

用蒙特卡羅方法估算 pi 值，核心方法是利用正方形和圓形面積的比例：

首先，我們在坐標(biāo)軸上構(gòu)造一個邊長為 1 的正方形

其次，我們以 (0, 0) 為圓心，構(gòu)造一個半徑為 1 的圓形

此時我們知道這個圓形有 1/4 是在正方形中的，正方形的面積和這 1/4 圓的面積分別是：1 和 pi/4，即 1/4 圓的面積和正方形面積之比剛好是 pi/4

然后通過蒙特卡羅模擬，看看這個比例大概是多少，模擬方法如下：

隨機(jī)扔 n 個點 (x, y)，其中 x, y 都在 0 和 1 之間

如果 x^2 + y^2 < 0，則把這個點標(biāo)注為紅色，表示這個點落在圓內(nèi)

最后數(shù)數(shù)有 n 個點中有多少點是紅點，即落在圓內(nèi)，假設(shè)點數(shù)為 m

則這個 1/4 圓的面積和正方形面積的比例應(yīng)該是：m/n，即 m/n = pi/4 => pi = 4*m/n

import numpy as np

def mc_pi(n=100):
    """Use Monte Calo Method to estimate pi.
    """
    m = 0
    i = 0
    while i < n:
        x, y = np.random.rand(2)
        if x**2 + y**2 < 1:
            m += 1
        i += 1

    pi = 4. * m / n
    res = {"total_point": n, "point_in_circle": m, "estimated_pi": pi}
    
    return res

我們按照上面寫的三大步驟來寫這個 spark 應(yīng)用：

加載數(shù)據(jù)集

### iterate number

total = int(100 * 10000)
local_collection = xrange(1, total)

### parallelize a data set into the cluster

rdd = sc.parallelize(local_collection)       
    .setName("parallelized_data")        
    .cache()

處理數(shù)據(jù)

### randomly generate points

def map_func(element):
    x = random.random()       ## [0, 1)
    y = random.random()       ## [0, 1)
    
    return (x, y)             ## random point

def map_func_2(element):
    x, y = element
    return 1 if x**2 + y**2 < 1 else 0

rdd2 = rdd.map(map_func)            
          .setName("random_point")  
          .cache()

### calculate the number of points in and out the circle

rdd3 = rdd2.map(map_func_2)                 
           .setName("points_in_out_circle") 
           .cache()

結(jié)果展示

### how many points are in the circle

in_circle = rdd3.reduce(operator.add)
pi = 4. * in_circle / total
print "iterate {} times".format(total)
print "estimated pi : {}".format(pi)

上面這個例子，可能會讓一些初步接觸 spark 的人很困惑，"明明幾行代碼就能解決的問題在 spark 里還有按照這些步驟寫這么多代碼？難道是老濕又騙我了嗎？"。

其實，就從上面這個例子看起來，似乎 spark 真的沒有什么優(yōu)勢，但是，上面這個例子的目的是表明 spark 的編程模式，如果你還不相信，可以把模擬次數(shù)加到千萬或者億次以上看看效果。

如果，如果你還是糾結(jié)于 "我騙了你，spark 沒有夢想中的那么好" 的話，那看下面這一行代碼吧，它也完成了同樣的事情：

### version 1
sc.parallelize(xrange(total))                                 
    .map(lambda x: (random.random(), random.random()))        
    .map(lambda x: 1 if x[0]**2 + x[1]**2 < 1 else 0)         
    .reduce(lambda x, y: x + y)                               
    / float(total) * 4

### version 2
sc.parallelize(xrange(total))                                  
    .map(lambda x: 1 if sum(np.random.random(2) ** 2) else 0)  
    .reduce(lambda x, y: x + y)                                
    / float(total) * 4

下一篇，介紹 spark 的 RDD，之后會多帶帶介紹 spark 的 dataframe 和 datasets。

[spark-rdd-paper : Resilient Distributed Datasets: A Fault-Tolerant Abstraction for
In-Memory Cluster Computing](http://litaotao.github.io/files/spark-rdd-paper.pdf)

spark python API

spark context API

機(jī)器學(xué)習(xí)相關(guān)數(shù)據(jù)集-斯坦福

spark pagerank example

latex online editor 在線latex公式編輯器

阮一峰：蒙特卡羅

蒙特卡羅，wikipedia

科學(xué)網(wǎng)：蒙特卡羅

『 Spark 』1. spark 簡介

『 Spark 』2. spark 基本概念解析

『 Spark 』3. spark 編程模式

『 Spark 』4. spark 之 RDD

『 Spark 』5. 這些年，你不能錯過的 spark 學(xué)習(xí)資源

『 Spark 』6. 深入研究 spark 運行原理之 job, stage, task

『 Spark 』7. 使用 Spark DataFrame 進(jìn)行大數(shù)據(jù)分析