ML Spark Pyspark

word count 實(shí)戰(zhàn)

用sc.parallelize創(chuàng)建一個(gè)基本的RDD

wordsList = ['cat', 'elephant', 'rat', 'rat', 'cat']
wordsRDD = sc.parallelize(wordsList, 4)
# Print out the type of wordsRDD
print type(wordsRDD)

#lambda 函數(shù)
pluralLambdaRDD = wordsRDD.map(lambda x : x+'s')
print pluralLambdaRDD.collect()

#Out:['cats', 'elephants', 'rats', 'rats', 'cats']

#用map 計(jì)算每個(gè)單詞的長(zhǎng)度
pluralLengths = (pluralRDD
                 .map(lambda x: len(x))
                 .collect())
print pluralLengths
#Out:[4, 9, 4, 4, 4]

創(chuàng)建一個(gè)pari RDD，值是 (k,v) k 是key v是value

wordPairs = wordsRDD.map(lambda x : (x,1))
print wordPairs.collect()

#Out: [('cat', 1), ('elephant', 1), ('rat', 1), ('rat', 1), ('cat', 1)]

用reduceByKey計(jì)算個(gè)數(shù)

# Note that reduceByKey takes in a function that accepts two values and returns a single value
wordCounts = wordPairs.reduceByKey(lambda x,y : x+y)
print wordCounts.collect()
#Out:[('rat', 2), ('elephant', 1), ('cat', 2)]

word count

#各步驟合并在一起 
wordCountsCollected = (wordsRDD
                       .map(lambda x : (x,1))
                       .reduceByKey(lambda x,y: x+y)
                       .collect())
print wordCountsCollected
#Out:[('rat', 2), ('elephant', 1), ('cat', 2)]

計(jì)算 均值

from operator import add
#去重后記錄條數(shù)
uniqueWords = wordCounts.count()
#總記錄條數(shù)
totalCount = (wordCounts
              .map(lambda (k,v): (v))
              .reduce(add))
#均值              
average = totalCount / float(uniqueWords)
print totalCount
#Out: 5
print round(average, 2)
#Out: 1.67

實(shí)戰(zhàn)##

1.定義wordCount函數(shù)，輸入wordListRDD，返回 wordCount RDD.

def wordCount(wordListRDD):
    """Creates a pair RDD with word counts from an RDD of words.
    Args:
        wordListRDD (RDD of str): An RDD consisting of words.
    Returns:
        RDD of (str, int): An RDD consisting of (word, count) tuples.
    """
    return (wordListRDD
            .map(lambda x :(x,1))
            .reduceByKey(lambda x,y:x+y)
            )
            
print wordCount(wordsRDD).collect()
#Out:[('rat', 2), ('elephant', 1), ('cat', 2)]

2.定義特殊字符處理函數(shù)

import re
def removePunctuation(text):
    """Removes punctuation, changes to lower case, and strips leading and trailing spaces.
    Note:
        Only spaces, letters, and numbers should be retained.  Other characters should should be
        eliminated (e.g. it's becomes its).  Leading and trailing spaces should be removed after
        punctuation is removed.
    Args:
        text (str): A string.
    Returns:
        str: The cleaned up string.
    """
    return re.sub(r'[^\w\s]','',text).strip().lower()

    
print removePunctuation('Hi, you!')
print removePunctuation(' No under_score!')
print removePunctuation(' *      Remove punctuation then spaces  * ')

#Out:hi you
#Out:no under_score
#Out:remove punctuation then spaces

3.導(dǎo)入文件

import os.path
baseDir = os.path.join('data')
inputPath = os.path.join('cs100', 'lab1', 'shakespeare.txt')
fileName = os.path.join(baseDir, inputPath)

#導(dǎo)入文件，處理特殊字符
shakespeareRDD = (sc
                  .textFile(fileName, 8)
                  .map(removePunctuation))
                  
                  
print '\n'.join(shakespeareRDD
                .zipWithIndex()  # to (line, lineNum)
                .map(lambda (l, num): '{0}: {1}'.format(num, l))  # to 'lineNum: line'
                .take(10))
"""
Out:
0: 1609
1: 
2: the sonnets
3: 
4: by william shakespeare
5: 
6: 
7: 
8: 1
9: from fairest creatures we desire increase
10: that thereby beautys rose might never die
"""

4.字符串轉(zhuǎn)成單詞

shakespeareWordsRDD = shakespeareRDD.flatMap(lambda x: x.split(' '))
shakespeareWordCount = shakespeareWordsRDD.count()
print shakespeareWordsRDD.top(5)
print shakespeareWordCount
#Out:[u'zwaggerd', u'zounds', u'zounds', u'zounds', u'zounds']
#Out:927631

5.過濾空字符

shakeWordsRDD = shakespeareWordsRDD.filter(lambda x: x<>'')

shakeWordCount = shakeWordsRDD.count()
print shakeWordCount
#Out:882996

6.計(jì)算Top15 單詞

top15WordsAndCounts = (wordCount(shakeWordsRDD)
                       .takeOrdered(15,key=lambda x: -1 * x[1] ))
print '\n'.join(map(lambda (w, c): '{0}: {1}'.format(w, c), top15WordsAndCounts))

"""
Out:
the: 27361
and: 26028
i: 20681
to: 19150
of: 17463
a: 14593
you: 13615
my: 12481
in: 10956
that: 10890
is: 9134
not: 8497
with: 7771
me: 7769
it: 7678
"""