pyspark-session-2019-05-09.txt

Learn Partitioning RDDs and using mapPartitions() Transformation

$ ./bin/pyspark
Python 3.7.2 (v3.7.2:9a3ffc0492, Dec 24 2018, 02:44:43)
[Clang 6.0 (clang-600.0.57)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /__ / .__/\_,_/_/ /_/\_\   version 2.4.0
      /_/

Using Python version 3.7.2 (v3.7.2:9a3ffc0492, Dec 24 2018 02:44:43)
SparkSession available as 'spark'.
>>> numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
>>>
>>>
>>> numbers
[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
>>> rdd = spark.sparkContext.parallelize(numbers, 3)
>>> rdd.count()
10
>>> rdd.collect()
[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
>>>

>>> def f(iterator):
...     for x in iterator:
...             print(x)
...     print("===")
...
>>>
>>> rdd.foreachPartition(f)
4
5
6
===
7
8
9
10
===
1
2
3
===
>>>
>>>
>>> rdd = spark.sparkContext.parallelize(numbers, 2)
>>> rdd.foreachPartition(f)
1
2
3
4
5
===
6
7
8
9
10
===
>>> 
>>> n = rdd.getNumPartitions()
>>> n
2
>>> rdd = spark.sparkContext.parallelize(numbers, 4)
>>> n = rdd.getNumPartitions()
>>> n
4
>>> rdd.foreachPartition(f)
5
6
===
3
4
===
7
8
9
10
===
1
2
===
>>> rdd = spark.sparkContext.parallelize(numbers, 14)
>>> rdd.foreachPartition(f)
4
===
===
===
3
===
1
===
5
===
2
===
===
6
===
===
8
===
7
===
9
===
10
===
>>> def min_max_count(iterator):
...     firsttime = 1
...     #minimum
...     #maximum
...     #count
...     for x in iterator:
...             if (firsttime == 1):
...                     minimum = x
...                     maximum = x
...                     count = 1
...                     firsttime = 0
...             else:
...                     count = count + 1
...                     minimum = min(x, minimum)
...                     maximum = max(x, maximum)
...             #
...     return (minimum, maximum, count)
...
>>>
>>> data = [12, 34, 3, 5, 7, 9, 91, 77, 12, 13, 14, 15, 16]
>>> data
[12, 34, 3, 5, 7, 9, 91, 77, 12, 13, 14, 15, 16]
>>> rdd = spark.sparkContext.parallelize(numbers, 3)
>>> n = rdd.getNumPartitions()
>>> n
3
>>> rdd.collect()
[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
>>> result = rdd.mapPartitions(min_max_count)
>>> result.collect()
[1, 3, 3, 4, 6, 3, 7, 10, 4]
>>> def min_max_count(iterator):
...     firsttime = 1
...     #minimum
...     #maximum
...     #count
...     for x in iterator:
...             if (firsttime == 1):
...                     minimum = x
...                     maximum = x
...                     count = 1
...                     firsttime = 0
...             else:
...                     count = count + 1
...                     minimum = min(x, minimum)
...                     maximum = max(x, maximum)
...             #
...     return [minimum, maximum, count]
...
>>>
>>> result = rdd.mapPartitions(min_max_count)
>>> result.collect()
[1, 3, 3, 4, 6, 3, 7, 10, 4]
>>>
>>>
>>>
>>>
>>> def min_max_count(iterator):
...     firsttime = 1
...     #minimum
...     #maximum
...     #count
...     for x in iterator:
...             if (firsttime == 1):
...                     minimum = x
...                     maximum = x
...                     count = 1
...                     firsttime = 0
...             else:
...                     count = count + 1
...                     minimum = min(x, minimum)
...                     maximum = max(x, maximum)
...             #
...     return [[minimum, maximum, count]]
...
>>> result = rdd.mapPartitions(min_max_count)
>>> result.collect()
[[1, 3, 3], [4, 6, 3], [7, 10, 4]]
>>>

>>> data
[12, 34, 3, 5, 7, 9, 91, 77, 12, 13, 14, 15, 16]
>>> rdd = spark.sparkContext.parallelize(data, 3)
>>>
>>>
>>> result = rdd.mapPartitions(min_max_count)
>>> result.collect()
[[3, 34, 4], [7, 91, 4], [12, 16, 5]]
>>> rdd.foreachPartition(f)
12
13
14
15
16
===
7
9
91
77
===
12
34
3
5
===
>>>