以下代碼用 Pyspark + IPython 完成

創(chuàng)新互聯(lián)公司的客戶(hù)來(lái)自各行各業(yè)，為了共同目標(biāo)，我們?cè)诠ぷ魃厦芮信浜希瑥膭?chuàng)業(yè)型小企業(yè)到企事業(yè)單位，感謝他們對(duì)我們的要求，感謝他們從不同領(lǐng)域給我們帶來(lái)的挑戰(zhàn)，讓我們激情的團(tuán)隊(duì)有機(jī)會(huì)用頭腦與智慧不斷的給客戶(hù)帶來(lái)驚喜。專(zhuān)業(yè)領(lǐng)域包括做網(wǎng)站、網(wǎng)站建設(shè)、電商網(wǎng)站開(kāi)發(fā)、微信營(yíng)銷(xiāo)、系統(tǒng)平臺(tái)開(kāi)發(fā)。

統(tǒng)計(jì)日志空行的數(shù)量：

讀取日志，創(chuàng)建RDD：

myrdd = sc.textFile("access.log")

1. 不使用累加器：

In [68]: s = 0
In [69]: def f(x):
    ...:     global s
    ...:     if len(x) == 0:
    ...:         s += 1
    ...:
In [70]: myrdd.foreach(f)
In [71]: print (s)

得出結(jié)果為：

0

原因是python 的變量，即使是全局變量不能應(yīng)用在各個(gè)計(jì)算進(jìn)程（線(xiàn)程）中同步數(shù)據(jù)，所以需要分布式計(jì)算框架的變量來(lái)同步數(shù)據(jù)，Spark 中采用累加器來(lái)解決：

2. 使用累加器

In [64]: s = sc.accumulator(0)
In [65]: def f(x):
    ...:     global s
    ...:     if len(x) == 0:
    ...:         s += 1
    ...:
In [66]: myrdd.foreach(f)
In [67]: print (s)

得出正確結(jié)果：

14

網(wǎng)站標(biāo)題：Spark累加器實(shí)驗(yàn)
轉(zhuǎn)載源于：http://www.aaarwkj.com/article18/pdisdp.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供網(wǎng)站排名、網(wǎng)頁(yè)設(shè)計(jì)公司、動(dòng)態(tài)網(wǎng)站、定制網(wǎng)站、網(wǎng)站收錄、網(wǎng)站維護(hù)

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶(hù)投稿、用戶(hù)轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請(qǐng)盡快告知，我們將會(huì)在第一時(shí)間刪除。文章觀(guān)點(diǎn)不代表本網(wǎng)站立場(chǎng)，如需處理請(qǐng)聯(lián)系客服。電話(huà)：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來(lái)源： 創(chuàng)新互聯(lián)