Jaka jest różnica między mapą a flatMap i dobrym przypadkiem użycia dla każdego z nich?

Czy ktoś może mi wyjaśnić różnicę między mapą a płaską mapą i jaki jest dobry przypadek użycia dla każdego z nich?

Co oznacza „spłaszczanie wyników”? Do czego to jest dobre?

Oto przykład różnicy jako spark-shellsesji:

Po pierwsze, niektóre dane - dwa wiersze tekstu:

val rdd = sc.parallelize(Seq("Roses are red", "Violets are blue"))  // lines

rdd.collect

    res0: Array[String] = Array("Roses are red", "Violets are blue")

Teraz mapprzekształca RDD o długości N w inny RDD o długości N.

Na przykład odwzorowuje dwie linie na dwie długości:

rdd.map(_.length).collect

    res1: Array[Int] = Array(13, 16)

Ale flatMap(luźno mówiąc) przekształca RDD o długości N w zbiór N kolekcji, a następnie spłaszcza je w pojedyncze RDD wyników.

rdd.flatMap(_.split(" ")).collect

    res2: Array[String] = Array("Roses", "are", "red", "Violets", "are", "blue")

Mamy wiele słów w wierszu i wiele wierszy, ale otrzymujemy jedną tablicę wyjściową słów

Aby to zilustrować, flatMapping z kolekcji linii do kolekcji słów wygląda następująco:

["aa bb cc", "", "dd"] => [["aa","bb","cc"],[],["dd"]] => ["aa","bb","cc","dd"]

Wejściowe i wyjściowe RDD będą zatem zazwyczaj różnych rozmiarów flatMap.

Gdybyśmy próbowali skorzystać mapz naszej splitfunkcji, mielibyśmy skończyło się struktur zagnieżdżonych (RDD tablic słów, z rodzaju RDD[Array[String]]), ponieważ musimy mieć dokładnie jeden wynik na wejściu:

rdd.map(_.split(" ")).collect

    res3: Array[Array[String]] = Array(
                                     Array(Roses, are, red), 
                                     Array(Violets, are, blue)
                                 )

Wreszcie jednym przydatnym przypadkiem specjalnym jest mapowanie za pomocą funkcji, która może nie zwrócić odpowiedzi, a zatem zwraca an Option. Możemy użyć flatMapdo odfiltrowania zwracanych elementów Nonei wyodrębnienia wartości z tych, które zwracają Some:

val rdd = sc.parallelize(Seq(1,2,3,4))

def myfn(x: Int): Option[Int] = if (x <= 2) Some(x * 10) else None

rdd.flatMap(myfn).collect

    res3: Array[Int] = Array(10,20)

(zauważając tutaj, że Opcja zachowuje się raczej jak lista zawierająca jeden element lub zero elementów)

Ogólnie używamy przykładu liczby słów w hadoopie. Wezmę ten sam przypadek użycia i użyje mapi flatMapi widzimy różnicę w jaki sposób przetwarza dane.

Poniżej znajduje się przykładowy plik danych.

hadoop is fast
hive is sql on hdfs
spark is superfast
spark is awesome

Powyższy plik zostanie przeanalizowany przy użyciu mapi flatMap.

Za pomocą map

>>> wc = data.map(lambda line:line.split(" "));
>>> wc.collect()
[u'hadoop is fast', u'hive is sql on hdfs', u'spark is superfast', u'spark is awesome']

Dane wejściowe mają 4 linie, a wielkość wyjściowa również 4, tj. N elementów ==> N elementów.

Za pomocą flatMap

>>> fm = data.flatMap(lambda line:line.split(" "));
>>> fm.collect()
[u'hadoop', u'is', u'fast', u'hive', u'is', u'sql', u'on', u'hdfs', u'spark', u'is', u'superfast', u'spark', u'is', u'awesome']

Wyjście różni się od mapy.

Przypiszmy 1 jako wartość dla każdego klawisza, aby uzyskać liczbę słów.

• fm: RDD utworzone przy użyciu flatMap
• wc: RDD utworzono za pomocą map

>>> fm.map(lambda word : (word,1)).collect()
[(u'hadoop', 1), (u'is', 1), (u'fast', 1), (u'hive', 1), (u'is', 1), (u'sql', 1), (u'on', 1), (u'hdfs', 1), (u'spark', 1), (u'is', 1), (u'superfast', 1), (u'spark', 1), (u'is', 1), (u'awesome', 1)]

Podczas gdy flatMapna RDD wcda poniższe niepożądane wyjście:

>>> wc.flatMap(lambda word : (word,1)).collect()
[[u'hadoop', u'is', u'fast'], 1, [u'hive', u'is', u'sql', u'on', u'hdfs'], 1, [u'spark', u'is', u'superfast'], 1, [u'spark', u'is', u'awesome'], 1]

Nie można uzyskać liczby słów, jeśli mapjest używany zamiast flatMap.

Zgodnie z definicją różnica między mapi flatMapwynosi:

map: Zwraca nowy RDD poprzez zastosowanie danej funkcji do każdego elementu RDD. Funkcja w mapzwraca tylko jeden element.

flatMap: Podobnie map, zwraca nowy RDD poprzez zastosowanie funkcji do każdego elementu RDD, ale dane wyjściowe są spłaszczone.

Jeśli pytasz o różnicę między RDD.map i RDD.flatMap w Spark, map przekształca RDD o rozmiarze N na inny o rozmiarze N. na przykład.

myRDD.map(x => x*2)

na przykład, jeśli myRDD składa się z Doubles.

Chociaż flatMap może przekształcić RDD w inny o innym rozmiarze: np .:

myRDD.flatMap(x =>new Seq(2*x,3*x))

który zwróci RDD o rozmiarze 2 * N lub

myRDD.flatMap(x =>if x<10 new Seq(2*x,3*x) else new Seq(x) )

Sprowadza się do twojego początkowego pytania: co rozumiesz przez spłaszczanie ?

Kiedy używasz flatMap, kolekcja „wielowymiarowa” staje się kolekcją „jednowymiarową” .

val array1d = Array ("1,2,3", "4,5,6", "7,8,9")  
//array1d is an array of strings

val array2d = array1d.map(x => x.split(","))
//array2d will be : Array( Array(1,2,3), Array(4,5,6), Array(7,8,9) )

val flatArray = array1d.flatMap(x => x.split(","))
//flatArray will be : Array (1,2,3,4,5,6,7,8,9)

Chcesz użyć płaskiej mapy, gdy,

• funkcja mapy powoduje tworzenie struktur wielowarstwowych
• ale wszystko, czego chcesz, to prosta - płaska - jednowymiarowa struktura, poprzez usunięcie WSZYSTKICH wewnętrznych zgrupowań

Użyj test.mdjako przykładu:

➜  spark-1.6.1 cat test.md
This is the first line;
This is the second line;
This is the last line.

scala> val textFile = sc.textFile("test.md")
scala> textFile.map(line => line.split(" ")).count()
res2: Long = 3

scala> textFile.flatMap(line => line.split(" ")).count()
res3: Long = 15

scala> textFile.map(line => line.split(" ")).collect()
res0: Array[Array[String]] = Array(Array(This, is, the, first, line;), Array(This, is, the, second, line;), Array(This, is, the, last, line.))

scala> textFile.flatMap(line => line.split(" ")).collect()
res1: Array[String] = Array(This, is, the, first, line;, This, is, the, second, line;, This, is, the, last, line.)

Jeśli użyjesz mapmetody, otrzymasz wiersze test.md, dlaflatMap metody, otrzymasz liczbę słów.

mapMetoda jest podobna do flatMap, wszystkie są nowe powrót RDD. mapmetoda często używać zwraca nowy RDD, flatMapmetoda często używa podzielonych słów.

mapzwraca RDD równej liczbie elementów, a flatMapmoże nie.

Przykład zastosowania przypadkuflatMap odfiltrowania brakujących lub niepoprawnych danych.

Przykładowy przypadek użycia domap użycia w wielu różnych przypadkach, w których liczba elementów wejściowych i wyjściowych jest taka sama.

liczba. cv

1
2
3
-
4
-
5

map.py dodaje wszystkie liczby w add.csv.

from operator import *

def f(row):
  try:
    return float(row)
  except Exception:
    return 0

rdd = sc.textFile('a.csv').map(f)

print(rdd.count())      # 7
print(rdd.reduce(add))  # 15.0

flatMap.py używa flatMapdo filtrowania brakujących danych przed dodaniem. Dodano mniej liczb w porównaniu do poprzedniej wersji.

from operator import *

def f(row):
  try:
    return [float(row)]
  except Exception:
    return []

rdd = sc.textFile('a.csv').flatMap(f)

print(rdd.count())      # 5
print(rdd.reduce(add))  # 15.0

map i flatMap są podobne, w tym sensie, że pobierają linię z wejściowego RDD i stosują na niej funkcję. Różnią się one tym, że funkcja w mapie zwraca tylko jeden element, podczas gdy funkcja w flatMap może zwrócić listę elementów (0 lub więcej) jako iterator.

Również wyjście flatMap jest spłaszczone. Chociaż funkcja w flatMap zwraca listę elementów, flatMap zwraca RDD, która ma wszystkie elementy z listy w sposób płaski (nie listę).

wszystkie przykłady są dobre .... Oto ładna wizualna ilustracja ... źródło dzięki uprzejmości: szkolenie iskier DataFlair

Mapa: Mapa to operacja transformacji w Apache Spark. Odnosi się do każdego elementu RDD i zwraca wynik jako nowy RDD. Na mapie programista operacji może zdefiniować własną logikę biznesową. Ta sama logika zostanie zastosowana do wszystkich elementów RDD.

mapFunkcja Spark RDD przyjmuje jeden element jako proces wejściowy zgodnie z niestandardowym kodem (określonym przez programistę) i zwraca jeden element na raz. Mapa przekształca RDD o długości N w inny RDD o długości N. Wejściowe i wyjściowe RDD będą zazwyczaj miały tę samą liczbę rekordów.

Przykład mapużycia scala:

val x = spark.sparkContext.parallelize(List("spark", "map", "example",  "sample", "example"), 3)
val y = x.map(x => (x, 1))
y.collect
// res0: Array[(String, Int)] = 
//    Array((spark,1), (map,1), (example,1), (sample,1), (example,1))

// rdd y can be re writen with shorter syntax in scala as 
val y = x.map((_, 1))
y.collect
// res1: Array[(String, Int)] = 
//    Array((spark,1), (map,1), (example,1), (sample,1), (example,1))

// Another example of making tuple with string and it's length
val y = x.map(x => (x, x.length))
y.collect
// res3: Array[(String, Int)] = 
//    Array((spark,5), (map,3), (example,7), (sample,6), (example,7))

FlatMap:

A flatMapjest operacją transformacji. Odnosi się do każdego elementu RDD i zwraca wynik jako nowy RDD. Jest podobny do mapy, ale FlatMap pozwala zwrócić 0, 1 lub więcej elementów z funkcji mapy. W operacji FlatMap programista może zdefiniować własną logikę biznesową. Ta sama logika zostanie zastosowana do wszystkich elementów RDD.

Co oznacza „spłaszczanie wyników”?

Funkcja FlatMap pobiera jeden element jako proces wejściowy zgodnie z niestandardowym kodem (określonym przez programistę) i zwraca 0 lub więcej elementów jednocześnie. flatMap() przekształca RDD o długości N w inny RDD o długości M.

Przykład flatMapużycia scala:

val x = spark.sparkContext.parallelize(List("spark flatmap example",  "sample example"), 2)

// map operation will return Array of Arrays in following case : check type of res0
val y = x.map(x => x.split(" ")) // split(" ") returns an array of words
y.collect
// res0: Array[Array[String]] = 
//  Array(Array(spark, flatmap, example), Array(sample, example))

// flatMap operation will return Array of words in following case : Check type of res1
val y = x.flatMap(x => x.split(" "))
y.collect
//res1: Array[String] = 
//  Array(spark, flatmap, example, sample, example)

// RDD y can be re written with shorter syntax in scala as 
val y = x.flatMap(_.split(" "))
y.collect
//res2: Array[String] = 
//  Array(spark, flatmap, example, sample, example)

Różnicę widać poniżej przykładowego kodu pyspark:

rdd = sc.parallelize([2, 3, 4])
rdd.flatMap(lambda x: range(1, x)).collect()
Output:
[1, 1, 2, 1, 2, 3]


rdd.map(lambda x: range(1, x)).collect()
Output:
[[1], [1, 2], [1, 2, 3]]

Flatmap i Map przekształcają kolekcję.

Różnica:

map (func)
Zwraca nowy rozproszony zestaw danych utworzony przez przepuszczenie każdego elementu źródła przez funkcję func.

flatMap (func)
Podobne do mapy, ale każdy element wejściowy może być odwzorowany na 0 lub więcej elementów wyjściowych (więc func powinien zwrócić Seq, a nie pojedynczy element).

Funkcja transformacji:
mapa : jeden element na -> jeden element na zewnątrz.
flatMap : Jeden element na -> 0 lub więcej elementów na zewnątrz (kolekcja).

RDD.map zwraca wszystkie elementy w jednej tablicy

RDD.flatMap zwraca elementy w tablicach tablicy

załóżmy, że mamy tekst w pliku text.txt jako

Spark is an expressive framework
This text is to understand map and faltMap functions of Spark RDD

Korzystanie z mapy

val text=sc.textFile("text.txt").map(_.split(" ")).collect

wynik:

text: **Array[Array[String]]** = Array(Array(Spark, is, an, expressive, framework), Array(This, text, is, to, understand, map, and, faltMap, functions, of, Spark, RDD))

Korzystanie z flatMap

val text=sc.textFile("text.txt").flatMap(_.split(" ")).collect

wynik:

 text: **Array[String]** = Array(Spark, is, an, expressive, framework, This, text, is, to, understand, map, and, faltMap, functions, of, Spark, RDD)

Dla wszystkich, którzy chcieli powiązać PySpark:

Przykładowa transformacja: flatMap

>>> a="hello what are you doing"
>>> a.split()

['cześć co robisz']

>>> b=["hello what are you doing","this is rak"]
>>> b.split()

Traceback (ostatnie ostatnie wywołanie): Plik „”, wiersz 1, w AttributeError: obiekt „list” nie ma atrybutu „split”

>>> rline=sc.parallelize(b)
>>> type(rline)

>>> def fwords(x):
...     return x.split()


>>> rword=rline.map(fwords)
>>> rword.collect()

[[„hello”, „what”, „are”, „you”, „doing”], [„this”, „is”, „rak”]]

>>> rwordflat=rline.flatMap(fwords)
>>> rwordflat.collect()

[„hello”, „what”, „are”, „you”, „doing”, „this”, „is”, „rak”]

Mam nadzieję, że to pomoże :)

map: Zwraca nowy RDDprzez zastosowanie funkcji do każdego elementu RDD. Funkcja w .map może zwrócić tylko jeden element.

flatMap: Podobnie do mapy, zwraca nową RDDpoprzez zastosowanie funkcji do każdego elementu RDD, ale dane wyjściowe są spłaszczone.

Również funkcja in flatMapmoże zwrócić listę elementów (0 lub więcej)

Na przykład:

sc.parallelize([3,4,5]).map(lambda x: range(1,x)).collect()

Wyjście: [[1, 2], [1, 2, 3], [1, 2, 3, 4]]

sc.parallelize([3,4,5]).flatMap(lambda x: range(1,x)).collect()

Wyjście: informacja o / p jest spłaszczona na jednej liście [1, 2, 1, 2, 3, 1, 2, 3, 4]

Źródło: https://www.linkedin.com/pulse/difference-between-map-flatmap-transformations-spark-pyspark-pandey/

mapa:

jest metodą wyższego rzędu, która przyjmuje funkcję jako dane wejściowe i stosuje ją do każdego elementu w źródłowym RDD.

http://commandstech.com/difference-between-map-and-flatmap-in-spark-what-is-map-and-flatmap-with-examples/

flatMap:

metoda wyższego rzędu i operacja transformacji, która przyjmuje funkcję wejściową.

Różnica w wynikach map i flatMap:

1.flatMap

val a = sc.parallelize(1 to 10, 5)

a.flatMap(1 to _).collect()

Wynik:

 1, 1, 2, 1, 2, 3, 1, 2, 3, 4, 1, 2, 3, 4, 5, 1, 2, 3, 4, 5, 6, 1, 2, 3, 4, 5, 6, 7, 1, 2, 3, 4, 5, 6, 7, 8, 1, 2, 3, 4, 5, 6, 7, 8, 9, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10

2 map.:

val a = sc.parallelize(List("dog", "salmon", "salmon", "rat", "elephant"), 3)

val b = a.map(_.length).collect()

Wynik:

3 6 6 3 8

• map (func) Zwraca nowy rozproszony zestaw danych utworzony przez przepuszczenie każdego elementu źródła przez zadeklarowaną funkcję func. więc map () to pojedynczy termin

whiles

• flatMap (func) Podobne do mapy, ale każdy element wejściowy może być odwzorowany na 0 lub więcej elementów wyjściowych, więc func powinien zwrócić Sekwencję zamiast pojedynczego elementu.