PYSPARK RDD - Transformacje

W Pythonie PySpark jest modułem iskry używanym do zapewnienia podobnego rodzaju przetwarzania, takiego jak Spark.

RDD oznacza odporne rozproszone zestawy danych. Możemy nazwać RDD podstawową strukturą danych w Apache Spark.

Musimy zaimportować RDD z Pyspark.Moduł RDD.

Więc w PYSPARK, aby utworzyć RDD, możemy użyć metody równoległości ().

Składnia:

Spark_App.SparkContext.równoległość (dane)

Gdzie,

Dane mogą być jednym wymiarami (dane liniowe) lub danych dwuwymiarowych (dane z kolumny wierszy).

Transformacje RDD:

RDD transformacji to operacja stosowana do RDD w celu tworzenia nowych danych z istniejącego RDD. Korzystając z transformacji, jesteśmy w stanie filtrować RDD, stosując niektóre transformacje.

Zobaczmy transformacje przeprowadzane na podanym RDD.

Omówimy je jeden po drugim.

1. mapa()

Transformacja map () służy do mapowania wartości do elementów obecnych w RDD. Wymaga anonimowej funkcji jako parametru, takiej jak Lambda i przekształca elementy w RDD.

Składnia:

Rdd_data.mapa (anonimowa czynność)

Parametry:

Anonimowa funkcja wygląda jak:

Element Lambda: Operacja

Na przykład operacja polega na dodaniu/odjęciu wszystkich elementów za pomocą nowego elementu.

Zobaczmy przykłady, aby lepiej zrozumieć tę transformację.

Przykład 1:

W tym przykładzie tworzymy RDD o nazwie Student_marks z 20 elementami i stosujemy transformację map (), dodając każdy element z 20 i wyświetlając je za pomocą akcji Collect ().

#Wportu na moduł PYSPARK
Import Pyspark
#Import Sparksession do tworzenia sesji
od Pyspark.SQL Import Sparksession
# Importuj RDD z Pyspark.RDD
od Pyspark.RDD Import RDD
#Utworz aplikację o nazwie Linuxhint
Spark_App = Sparksession.budowniczy.AppName („Linuxhint”).getorCreate ()
# Utwórz dane uczniów z 20 elementami
student_marks = Spark_App.SparkContext.równoległe ([89,76,78,89,90 100, 34,56,54,22,45,43,23,56,78,21,34,34,56,34])
#Display Dane w RDD
Drukuj („Rzeczywiste dane w RDD:”, Student_marks.Mapa (element Lambda: Element).zbierać())
#Apply Map () Transformacja poprzez dodanie 20 do każdego elementu w RDD
Drukuj („Po dodaniu 20 do każdego elementu w RDD:”, Student_marks.Mapa (element Lambda: element+ 20).zbierać())

Wyjście:

Rzeczywiste dane w RDD: [89, 76, 78, 89, 90, 100, 34, 56, 54, 22, 45, 43, 23, 56, 78, 21, 34, 34, 56, 34]
Po dodaniu 20 do każdego elementu w RDD: [109, 96, 98, 109, 110, 120, 54, 76, 74, 42, 65, 63, 43, 76, 98, 41, 54, 54, 76, 54]

Z powyższego wyjścia widać, że element 20 jest dodawany do każdego elementu w RDD poprzez funkcję Lambda za pomocą transformacji map ().

Przykład 2:

W tym przykładzie tworzymy RDD o nazwie Student_marks z 20 elementami i stosujemy transformację map (), odejmując każdy element przez 15 i wyświetlanie go za pomocą akcji zbiorowej ().

#Wportu na moduł PYSPARK
Import Pyspark
#Import Sparksession do tworzenia sesji
od Pyspark.SQL Import Sparksession
# Importuj RDD z Pyspark.RDD
od Pyspark.RDD Import RDD
#Utworz aplikację o nazwie Linuxhint
Spark_App = Sparksession.budowniczy.AppName („Linuxhint”).getorCreate ()
# Utwórz dane uczniów z 20 elementami
student_marks = Spark_App.SparkContext.równoległe ([89,76,78,89,90 100, 34,56,54,22,45,43,23,56,78,21,34,34,56,34])
#Display Dane w RDD
Drukuj („Rzeczywiste dane w RDD:”, Student_marks.Mapa (element Lambda: Element).zbierać())
#Apply Map () Transformacja przez odejmowanie 15 od każdego elementu w RDD
Drukuj („Po odjęciu 15 z każdego elementu w RDD:”, Student_marks.Mapa (element Lambda: element 15).zbierać())

Wyjście:

Rzeczywiste dane w RDD: [89, 76, 78, 89, 90, 100, 34, 56, 54, 22, 45, 43, 23, 56, 78, 21, 34, 34, 56, 34]
Po odjęciu 15 od każdego elementu w RDD: [74, 61, 63, 74, 75, 85, 19, 41, 39, 7, 30, 28, 8, 41, 63, 6, 19, 19, 41, 19]

Z powyższego wyjścia widzimy, że element 15 jest odejmowany do każdego elementu w RDD poprzez funkcję Lambda za pomocą transformacji map ().

2. filtr()

transformacja filtr () służy do filtrowania wartości z RDD. Podejmuje anonimową funkcję, taką jak Lambda i zwraca elementy, filtrując elementy z RDD.

Składnia:

Rdd_data.Filtr (Anonimowy_funkcja)

Parametry:

Anonimowa funkcja wygląda jak:

Element Lambda: Warunek/wyrażenie

Na przykład warunek służy do określenia instrukcji ekspresyjnych w celu filtrowania RDD.

Zobaczmy przykłady, aby lepiej zrozumieć tę transformację.

Przykład 1:

W tym przykładzie tworzymy RDD o nazwie Student_marks z 20 elementami i stosujemy transformację filtr (), filtrując tylko wielokrotności 5 i wyświetlanie ich za pomocą akcji COBOLD ().

#Wportu na moduł PYSPARK
Import Pyspark
#Import Sparksession do tworzenia sesji
od Pyspark.SQL Import Sparksession
# Importuj RDD z Pyspark.RDD
od Pyspark.RDD Import RDD
#Utworz aplikację o nazwie Linuxhint
Spark_App = Sparksession.budowniczy.AppName („Linuxhint”).getorCreate ()
# Utwórz dane uczniów z 20 elementami
student_marks = Spark_App.SparkContext.równoległe ([89,76,78,89,90 100, 34,56,54,22,45,43,23,56,78,21,34,34,56,34])
#Display Dane w RDD
Drukuj („Rzeczywiste dane w RDD:”, Student_marks.Mapa (element Lambda: Element).zbierać())
#Apply Filter () Transformacja poprzez zwracanie inly wielokrotności 5.
Drukuj („Mnożniki 5 z RDD:”, Student_marks.Filtr (element Lambda: element%5 == 0).zbierać())
)

Wyjście:

Rzeczywiste dane w RDD: [89, 76, 78, 89, 90, 100, 34, 56, 54, 22, 45, 43, 23, 56, 78, 21, 34, 34, 56, 34]
Mnożniki 5 z RDD: [90, 100, 45]

Z powyższego wyjścia widać, że wielokrotności 5 elementów są filtrowane z RDD.

Przykład 2:

W tym przykładzie tworzymy RDD o nazwie Student_marks z 20 elementami i stosujemy transformację filtr () poprzez filtrowanie elementów większych niż 45 i wyświetlanie ich za pomocą akcji zbiorów ().

#Wportu na moduł PYSPARK
Import Pyspark
#Import Sparksession do tworzenia sesji
od Pyspark.SQL Import Sparksession
# Importuj RDD z Pyspark.RDD
od Pyspark.RDD Import RDD
#Utworz aplikację o nazwie Linuxhint
Spark_App = Sparksession.budowniczy.AppName („Linuxhint”).getorCreate ()
# Utwórz dane uczniów z 20 elementami
student_marks = Spark_App.SparkContext.równoległe ([89,76,78,89,90 100, 34,56,54,22,45,43,23,56,78,21,34,34,56,34])
#Display Dane w RDD
Drukuj („Rzeczywiste dane w RDD:”, Student_marks.Mapa (element Lambda: Element).zbierać())
#Apply Filter () Transformacja poprzez filtrowanie wartości większych niż 45
Drukuj („Wartości większe niż 45:”, Student_marks.Filtr (element Lambda: element> 45).zbierać())

Wyjście:

Rzeczywiste dane w RDD: [89, 76, 78, 89, 90, 100, 34, 56, 54, 22, 45, 43, 23, 56, 78, 21, 34, 34, 56, 34]
Wartości większe niż 45: [89, 76, 78, 89, 90, 100, 56, 54, 56, 78, 56]

Z powyższego wyjścia widzimy, że te elementy większe niż 45 są filtrowane z RDD.

3. unia()

transformacja Union () służy do połączenia dwóch RDD. Możemy wykonać tę transformację na dwóch RDD…

Składnia:

Rdd_data1.Union (rdd_data2)

Zobaczmy przykłady, aby lepiej zrozumieć tę transformację.

Przykład 1:

W tym przykładzie utworzymy pojedynczy RDD z danymi znaków studenckich i wygenerujemy dwa RDD z pojedynczego RDD, filtrując niektóre wartości za pomocą transformacji Filter (). Następnie możemy przeprowadzić transformację Union () na dwóch przefiltrowanych RDD.

#Wportu na moduł PYSPARK
Import Pyspark
#Import Sparksession do tworzenia sesji
od Pyspark.SQL Import Sparksession
# Importuj RDD z Pyspark.RDD
od Pyspark.RDD Import RDD
#Utworz aplikację o nazwie Linuxhint
Spark_App = Sparksession.budowniczy.AppName („Linuxhint”).getorCreate ()
# Utwórz dane uczniów z 20 elementami
student_marks = Spark_App.SparkContext.równoległe ([89,76,78,89,90 100, 34,56,54,22,45,43,23,56,78,21,34,34,56,34])
#Display Dane w RDD
Drukuj („Rzeczywiste dane w RDD:”, Student_marks.Mapa (element Lambda: Element).zbierać())
First_filter = Student_marks.Filtr (element Lambda: element> 90)
Second_filter = Student_marks.Filtr (element Lambda: element <40)
#Display Pierwsza przefiltrowana transformacja
Drukuj („Elementy w RDD większe niż 90”, First_filter.zbierać())
#Display Druga przefiltrowana transformacja
Drukuj („Elementy w RDD mniej niż 40”, Second_filter.zbierać())
#Apply Union () Transformation poprzez wykonanie Unii na powyższych 2 filtrach
Drukuj („Transformacja Unii na dwóch przefiltrowanych danych”, First_filter.Union (Second_filter).zbierać())

Wyjście:

Rzeczywiste dane w RDD: [89, 76, 78, 89, 90, 100, 34, 56, 54, 22, 45, 43, 23, 56, 78, 21, 34, 34, 56, 34]
Elementy w RDD większe niż 90 [100]
Elementy w RDD mniej niż 40 [34, 22, 23, 21, 34, 34, 34]
Transformacja związkowa na dwóch przefiltrowanych danych [100, 34, 22, 23, 21, 34, 34, 34]

Z powyższego wyjścia widać, że wykonaliśmy Union na First_filter i Second_filter.

First_filter jest uzyskiwany przez uzyskanie elementów od uczniów RDD większe niż 90, a drugie filter uzyskuje się przez uzyskanie elementów od uczniów RDD mniej niż 40 za pomocą transformacji Filter ().

Przykład 2:

W tym przykładzie utworzymy dwa RDD, tak że pierwszy RDD ma 20 elementów, a drugi RDD ma 10 elementów. Następnie możemy zastosować transformację Union () do tych dwóch RDD.

#Wportu na moduł PYSPARK
Import Pyspark
#Import Sparksession do tworzenia sesji
od Pyspark.SQL Import Sparksession
# Importuj RDD z Pyspark.RDD
od Pyspark.RDD Import RDD
#Utworz aplikację o nazwie Linuxhint
Spark_App = Sparksession.budowniczy.AppName („Linuxhint”).getorCreate ()
# Utwórz dane uczniów z 20 elementami
Student_marks1 = Spark_App.SparkContext.równoległe ([89,76,78,89,90 100, 34,56,54,22,45,43,23,56,78,21,34,34,56,34])
# Utwórz dane uczniów z 10 elementami
student_marks2 = Spark_App.SparkContext.równoległość ([45,43,23,56,78,21,34,34,56,34]))
#Display Dane w RDD
Drukuj („Rzeczywiste dane w znakach studenckich 1 RDD:”, Student_marks1.Mapa (element Lambda: Element).zbierać())
#Display Dane w RDD
Drukuj („Rzeczywiste dane w znakach studenckich 2 RDD:”, Student_marks2.Mapa (element Lambda: Element).zbierać())
#Apply Union () Transformation poprzez wykonanie Unii na powyższych 2 RDD
Drukuj („Transformacja Unii na dwóch RDD”, student_marks1.Union (student_marks2).zbierać())

Wyjście:

Rzeczywiste dane w znakach studenckich 1 RDD: [89, 76, 78, 89, 90, 100, 34, 56, 54, 22, 45, 43, 23, 56, 78, 21, 34, 34, 56, 34]
Rzeczywiste dane w znakach studenckich 2 RDD: [45, 43, 23, 56, 78, 21, 34, 34, 56, 34]
Transformacja Unii na dwóch RDD [89, 76, 78, 89, 90, 100, 34, 56, 54, 22, 45, 43, 23, 56, 78, 21, 34, 34, 56, 34, 45, 43, 23, 56, 78, 21, 34, 34, 56, 34]

Widzimy, że dwa RDD są łączone przy użyciu transformacji Union ().

Wniosek

Z tego samouczka Pyspark widzimy trzy transformacje zastosowane do RDD. MAP () Transformacja służy do mapowania przez transformację elementów w RDD, Filter () służy do wykonywania operacji filtra i utworzenia nowego filtrowanego RDD z istniejącego RDD. Na koniec omówiliśmy Union () RDD, który służy do połączenia dwóch RDD.