Spark 核心概念与操作

import pyspark
from pyspark import SparkContext
from pyspark import SparkConf
conf=SparkConf().setAppName("miniProject").setMaster("local[*]")
sc=SparkContext.getOrCreate(conf)

my_list = [1,2,3,4,5] #存放在当前环境内存当中的list

rdd = sc.parallelize(my_list)
rdd
rdd.getNumPartitions()

# File from Pandas exercises
rdd = sc.textFile("../data/yob1880.txt")
rdd

# Record current path for future use
import os
cwd = os.getcwd()
cwd

# File from Pandas exercises
rdd = sc.textFile("file://" + cwd + "/names/yob1880.txt")
rdd

rdd = sc.wholeTextFiles("file://" + cwd + "/names")
rdd
# 取出第一行（tips：`rdd.first()`) 统计行数（tips：`rdd.count()`）
rdd.first()  

numbersRDD = sc.parallelize(range(1,10+1))
print(numbersRDD.collect())

squaresRDD = numbersRDD.map(lambda x: x**2)  # 1进1出
print(squaresRDD.collect())

filteredRDD = numbersRDD.filter(lambda x: x % 2 == 0)  # Only the evens
print(filteredRDD.collect())

sentencesRDD = sc.parallelize(['Hello world', 'My name is Patrick'])
wordsRDD = sentencesRDD.flatMap(lambda sentence: sentence.split(" "))
print(wordsRDD.collect())
print(wordsRDD.count())

l = ['Hello world', 'My name is Patrick']
ll = []
for sentence in l:
    ll = ll + sentence.split(" ")
ll

# [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
def doubleIfOdd(x):
    if x % 2 == 1:
        return 2 * x
    else:
        return x

resultRDD = (numbersRDD           # In parentheses so we can write each
             .map(doubleIfOdd)    # transformation in one line
             .filter(lambda x: x > 6)
             .distinct())

resultRDD.collect()

numbersRDD = sc.parallelize([1,2,3])
moreNumbersRDD = sc.parallelize([2,3,4])

numbersRDD.union(moreNumbersRDD).collect() # [1, 2, 3, 2, 3, 4]

numbersRDD.intersection(moreNumbersRDD).collect() # [2, 3]

numbersRDD.subtract(moreNumbersRDD).collect() # [1]

numbersRDD.cartesian(moreNumbersRDD).collect() # [(1, 2), (1, 3), (1, 4), (2, 2), (2, 3), (2, 4), (3, 2), (3, 3), (3, 4)]

# [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
rdd = sc.parallelize(range(1,10+1))
rdd.reduce(lambda x, y: x + y) # 55

def f(x,y):
    return x + y

l = [1,2,3,4]
f(f(f(l[0],l[1]), l[2]), l[3]) # 10

# Calculate the average of all the squares from 1 to 10
import numpy as np
numbersRDD = sc.parallelize(np.linspace(1.0, 10.0, 10))
squaresRDD = numbersRDD.map(lambda x: x**2)

squaresRDD.cache()  # Preserve the actual items of this RDD in memory

avg = squaresRDD.reduce(lambda x, y: x + y) / squaresRDD.count()
print(avg) # 38.5

rdd = sc.parallelize(["Hello hello", "Hello New York", "York says hello"])
resultRDD = (
    rdd
    .flatMap(lambda sentence: sentence.split(" "))  # split into words
    .map(lambda word: word.lower())                 # lowercase
    .map(lambda word: (word, 1))                    # count each appearance
    .reduceByKey(lambda x, y: x + y)                # add counts for each word
)
resultRDD.collect()  # [('hello', 4), ('york', 2), ('new', 1), ('says', 1)]

result = resultRDD.collectAsMap()
result # {'hello': 4, 'york': 2, 'new': 1, 'says': 1}

print(resultRDD
      .sortBy(keyfunc=lambda x: x[1], ascending=False)
      .take(2))  # [('hello', 4), ('york', 2)]

# Home of different people
homesRDD = sc.parallelize([
        ('Brussels', 'John'),
        ('Brussels', 'Jack'),
        ('Leuven', 'Jane'),
        ('Antwerp', 'Jill'),
    ])

# Quality of life index for various cities
lifeQualityRDD = sc.parallelize([
        ('Brussels', 10),
        ('Antwerp', 7),
        ('RestOfFlanders', 5),
    ])

'''[('Brussels', ('John', 10)),
 ('Brussels', ('Jack', 10)),
 ('Antwerp', ('Jill', 7))]'''
homesRDD.join(lifeQualityRDD).collect()

'''[('Brussels', ('John', 10)),
 ('Brussels', ('Jack', 10)),
 ('Antwerp', ('Jill', 7)),
 ('Leuven', ('Jane', None))]'''
homesRDD.leftOuterJoin(lifeQualityRDD).collect()

'''[('Brussels', ('John', 10)),
 ('Brussels', ('Jack', 10)),
 ('Antwerp', ('Jill', 7)),
 ('RestOfFlanders', (None, 5))]'''
homesRDD.rightOuterJoin(lifeQualityRDD).collect()

'''[('Brussels',
  (<pyspark.resultiterable.ResultIterable at 0x13ff8a30358>,
   <pyspark.resultiterable.ResultIterable at 0x13ff8a309e8>)),
 ('Antwerp',
  (<pyspark.resultiterable.ResultIterable at 0x13ff8a30e80>,
   <pyspark.resultiterable.ResultIterable at 0x13ff8a30ac8>)),
 ('RestOfFlanders',
  (<pyspark.resultiterable.ResultIterable at 0x13ff8a30b70>,
   <pyspark.resultiterable.ResultIterable at 0x13ff8a30f28>)),
 ('Leuven',
  (<pyspark.resultiterable.ResultIterable at 0x13ff8a302b0>,
   <pyspark.resultiterable.ResultIterable at 0x13ff8a30fd0>))]'''
homesRDD.cogroup(lifeQualityRDD).collect()

''' 
Oops!  Those <ResultIterable>s are Spark's way of returning a list
that we can walk over, without materializing the list.
Let's materialize the lists to make the above more readable:
输出结果：
[('Brussels', (['John', 'Jack'], [10])),
 ('Antwerp', (['Jill'], [7])),
 ('RestOfFlanders', ([], [5])),
 ('Leuven', (['Jane'], []))]
'''
(homesRDD
 .cogroup(lifeQualityRDD)
 .map(lambda x:
        (x[0], (list(x[1][0]), list(x[1][1]))))
 .collect())