from pyspark.sql import SparkSession
from pyspark import SparkContext, SparkConf
spark = SparkSession.builder.getOrCreate()
sc = spark.sparkContext
Using combineByKey()
>>> data = [('A', 2), ('A', 3), ('A', 4), ('A', 5), ('B', 6), ('B', 7), ('B', 8)]
>>> data
[('A', 2), ('A', 3), ('A', 4), ('A', 5), ('B', 6), ('B', 7), ('B', 8)]
>>> rdd = spark.sparkContext.parallelize(data)m
>>>
>>>
>>> rdd.count()
7
>>> rdd.collect()
[('A', 2), ('A', 3), ('A', 4), ('A', 5), ('B', 6), ('B', 7), ('B', 8)]
>>> # (K, (sum, count, min, max))
>>> def single(v):
return (v, 1, v, v)
>>> def merge(C, v):
return (C[0]+v, C[1]+1, min(C[2],v), max(C[3],v))
>>> def combine(C1, C2):
return (C1[0]+C2[0], C1[1]+C2[1], min(C1[2], C2[2]), max(C1[3], C2[3]))
>>> rdd2 = rdd.combineByKey(single, merge, combine)
>>> rdd2.collect()
[
('B', (21, 3, 6, 8)),
('A', (14, 4, 2, 5))
]
For the first value of each key, initialize a tuple (v, 10)
For the
<u># Using reduceByKey()</u>
data = [('A', 3), ('A', 4), ('A', -2), ('B', -3), ('B', 4), ('B', 5), ('C', 4), ('D', -7), ('D', -9), ('E', -5), ('F', -2), ('F', 0)]
# convert into rdd
rdd = sc.parallelize(data)
# convert (k, (v, 1)) pairs
count_data = rdd.mapValues(lambda x: (x,1))
>> [('A', (3, 1)), ('A', (4, 1)), ('A', (-2, 1)), ('B', (-3, 1)), ('B', (4, 1)), ('B', (5, 1)), ('C', (4, 1)), ('D', (-7, 1)), ('D', (-9, 1)), ('E', (-5, 1)), ('F', (-2, 1)), ('F', (0, 1))]
# Now use reduceByKey() to find the mean of the rdd
mean_reduce = count_data.reduceByKey(lambda x, y: (x[0]+y[0], x[1]+y[1])) \
.mapValues(lambda x: x[0]/x[1])
.collect()
>> [('B', 2.0), ('E', -5.0), ('F', -1.0), ('C', 4.0), ('D', -8.0), ('A', 1.6666666666666667)]
#--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
<u># Use groupByKey()</u>
data = [('A', 3), ('A', 4), ('A', -2), ('B', -3), ('B', 4), ('B', 5), ('C', 4), ('D', -7), ('D', -9), ('E', -5), ('F', -2), ('F', 0)]
# use groupByKey() to find the mean
mean_group = data.groupByKey().mapValues(lambda x: sum(x)/len(x))
mean_group.collect()
>> [('B', 2.0), ('E', -5.0), ('F', -1.0), ('C', 4.0), ('D', -8.0), ('A', 1.6666666666666667)]
employee_records = employee_lines.map(lambda x: x.split(','))\ .filter(lambda x: len(x) == 4)\ .map(lambda x: (x[0], (x[1], float(x[3]))))