Pyspark – podstawowe transformacje i akcje

Transformacje tworzą nowe obiekty RDD , również na podstawie innego obiektu RDD.
Silnik Apache Spark potrafi przeanalizować sekwencję transformacji.
Akcje wykonują na obiektach RDD działania, których wynikiem nie jest RDD i działania te wymuszają wykorzystanie definicji RDD wraz z sekwencją transformacji.

Proste przykłady w pySparku.

#modyfikacja konfiguracji pozwalajaca na wyswietlenie 
# wiecej niz jednego komunikatu OUT: z kodu jednej komorki
from IPython.core.interactiveshell import InteractiveShell
InteractiveShell.ast_node_interactivity = "all"

#pokaz obiekt SparkContext
sc

Out[2]:

# wyswietlenie typu zmiennej sc
type(sc)

Out[3]: pyspark.context.SparkContext

#tworzymy obiekt RDD  z listy elementów całkowitych
rdd = sc.parallelize([1, 2, 3, 4,5,6])
#pokaz 10 elementów
rdd.take (10)
#policz elementy obiektu RDD
rdd.count()

Out[4]:[1, 2, 3, 4, 5, 6]
Out[4]:6

#transformacja map za pomocą funkcji lambda
#liczymy element po elemencie i tworzymy nowy RDD
rddMap=rdd.map(lambda x: x*2)
#pokaz elementy rdd
rdd.take (10)
#pokaz elementy rddMap
rddMap.take (10)

Out[5]:[1, 2, 3, 4, 5, 6]

Out[5]:[2, 4, 6, 8, 10, 12]

#transformacja filter za pomocą funkcji lambda
# pozostaw tylko elementy parzyste
rddFilter=rdd.filter(lambda x: x % 2 == 0)
#pokaz elementy rdd
rdd.take(10)
#pokaz elementy rddFilter
rddFilter.take(10)

Out[6]:[1, 2, 3, 4, 5, 6]

Out[6]:[2, 4, 6]

#transformacja map za pomocą funkcji lambda
#zamiana w pary [elementx,elementx*2]
rddMap=rdd.map(lambda x: [x, x * 2])
#pokaz elementy rdd
rdd.take(10)
#pokaz elementy rddFilter
rddMap.take(10)

Out[7]: [1, 2, 3, 4, 5, 6]

Out[7]:[[1, 2], [2, 4], [3, 6], [4, 8], [5, 10], [6, 12]]

#transformacja flatMap za pomocą funkcji lambda
#kazdy elementx zamieniamy na dwa elementy elementx i elementx+2
rddFlatMap=rdd.flatMap(lambda x:[x,x+2])
#pokaz elementy rdd
rdd.take(10)
#pokaz elementy rdd
rddFlatMap.take(10)

Out[8]:[1, 2, 3, 4, 5, 6]

Out[8]:[1, 3, 2, 4, 3, 5, 4, 6, 5, 7]

#akcja reduce za pomocą funkcji lambda
rddReduce=rdd.reduce(lambda x, y: x * y)
#pokaz elementy rdd
rdd.take(10)
#wyswietl zmienna rddReduce
rddReduce

Out[9]:[1, 2, 3, 4, 5, 6]

Out[9]:720

# pozostaw tylko elementy parzyste
rddOdd=rdd.filter(lambda x: x % 2 == 0)
# pozostaw tylko elementy nieparzyste
rddEven=rdd.filter(lambda x: x % 2 == 1)
# połącz oba zniory transformacją union
rddAll=rddOdd.union(rddEven)
rddOdd.take(10)
rddEven.take(10)
rddAll.take(10)

Out[10]:[2, 4, 6]

Out[10]:[1, 3, 5]

Out[10]:[2, 4, 6, 1, 3, 5]

#transformacja distinct usuwajaca zdublowane wartosci
rddNotDistinct=sc.parallelize([1,2,3,4,5,6,3,4,5])
rddDistinct=rddNotDistinct.distinct()
rddNotDistinct.take(10)
rddDistinct.take(10)

Out[11]:[1, 2, 3, 4, 5, 6, 3, 4, 5]

Out[11]:[2, 4, 6, 1, 3, 5]

# pozostale transformacje union, intersection, subtract, cartesian
rddUnion=rddOdd.union(rddEven)
rddIntersection=rddOdd.intersection(rddEven)
rddSubtract=rddOdd.subtract(rddEven)
rddCartesian=rddOdd.cartesian(rddEven)
rddOdd.take(10)
rddEven.take(10)
rddUnion.take(10)
rddIntersection.take(10)
rddSubtract.take(10)
rddCartesian.take(10)

Out[12]:[2, 4, 6]

Out[12]:[1, 3, 5]

Out[12]:[2, 4, 6, 1, 3, 5]

Out[12]:[]

Out[12]:[4, 2, 6]

Out[12]:[(2, 1), (2, 3), (2, 5), (4, 1), (4, 3), (6, 1), (6, 3), (4, 5), (6, 5)]