¿Cuál es la diferencia entre map y flatMap y un buen caso de uso para cada uno?

¿Alguien puede explicarme la diferencia entre map y flatMap y cuál es un buen caso de uso para cada uno?

¿Qué significa "aplanar los resultados"? ¿Para que sirve?

Aquí hay un ejemplo de la diferencia, como spark-shellsesión:

Primero, algunos datos: dos líneas de texto:

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

rdd.collect

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

Ahora, maptransforma un RDD de longitud N en otro RDD de longitud N.

Por ejemplo, asigna de dos líneas a dos longitudes de línea:

rdd.map(_.length).collect

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

Pero flatMap(hablando en términos generales) transforma un RDD de longitud N en una colección de N colecciones, luego las aplana en un único RDD de resultados.

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

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

Tenemos varias palabras por línea y varias líneas, pero terminamos con una única matriz de palabras de salida

Solo para ilustrar eso, flatMapping de una colección de líneas a una colección de palabras se ve así:

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

Por lo tanto, los RDD de entrada y salida serán típicamente de diferentes tamaños para flatMap.

Si hubiéramos tratado de usar mapcon nuestra splitfunción, habríamos terminado con estructuras anidadas (un RDD de matrices de palabras, con tipo RDD[Array[String]]) porque tenemos que tener exactamente un resultado por entrada:

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

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

Finalmente, un caso especial útil es el mapeo con una función que podría no devolver una respuesta y, por lo tanto, devuelve un Option. Podemos usar flatMappara filtrar los elementos que devuelven Noney extraer los valores de los que devuelven un 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)

(observando aquí que una Opción se comporta más bien como una lista que tiene un elemento o cero elementos)

Generalmente usamos el ejemplo de conteo de palabras en hadoop. Tomaré el mismo caso de uso mapy usaré y flatMapveremos la diferencia en cómo se procesan los datos.

A continuación se muestra el archivo de datos de muestra.

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

El archivo anterior se analizará usando mapy flatMap.

Utilizando 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']

La entrada tiene 4 líneas y el tamaño de salida también es 4, es decir, N elementos ==> N elementos.

Utilizando 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']

El resultado es diferente del mapa.

Asignemos 1 como valor para cada clave para obtener el recuento de palabras.

• fm: RDD creado usando flatMap
• wc: RDD creado usando 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)]

Mientras que flatMapen RDD wcdará el siguiente resultado no deseado:

>>> 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]

No puede obtener el recuento de palabras si mapse usa en lugar de flatMap.

Según la definición, la diferencia entre mapy flatMapes:

map: Devuelve un nuevo RDD aplicando una función dada a cada elemento del RDD. La función en mapdevuelve solo un artículo.

flatMap: Similar a map, devuelve un nuevo RDD aplicando una función a cada elemento del RDD, pero la salida es plana.

Si está preguntando la diferencia entre RDD.map y RDD.flatMap en Spark, map transforma un RDD de tamaño N a otro de tamaño N. p.ej.

myRDD.map(x => x*2)

por ejemplo, si myRDD está compuesto por Dobles.

Si bien flatMap puede transformar el RDD en otro de un tamaño diferente: por ejemplo:

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

que devolverá un RDD de tamaño 2 * N o

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

Se reduce a su pregunta inicial: ¿qué quiere decir con aplanamiento ?

Cuando utiliza flatMap, una colección "multidimensional" se convierte en una colección "unidimensional" .

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)

Desea usar un flatMap cuando,

• la función del mapa da como resultado la creación de estructuras de varias capas
• pero todo lo que quiere es una estructura simple, plana y unidimensional, eliminando TODAS las agrupaciones internas

Usar test.mdcomo ejemplo:

➜  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.)

Si usa el mapmétodo, obtendrá las líneas de test.md, para el flatMapmétodo, obtendrá el número de palabras.

El mapmétodo es similar a flatMap, todos ellos devuelven un nuevo RDD. mapmétodo a menudo para usar devolver un nuevo RDD, flatMapmétodo a menudo para usar palabras divididas.

mapdevuelve RDD del mismo número de elementos, mientras que flatMappuede que no.

Un ejemplo de caso de uso paraflatMap Filtrar datos faltantes o incorrectos.

Un ejemplo de caso demap uso para Uso en una amplia variedad de casos donde el número de elementos de entrada y salida es el mismo.

number.csv

1
2
3
-
4
-
5

map.py agrega todos los números en 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 utiliza flatMappara filtrar los datos faltantes antes de la adición. Se agregan menos números en comparación con la versión anterior.

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 y flatMap son similares, en el sentido de que toman una línea del RDD de entrada y le aplican una función. La forma en que difieren es que la función en el mapa devuelve solo un elemento, mientras que la función en flatMap puede devolver una lista de elementos (0 o más) como un iterador.

Además, la salida de flatMap se aplana. Aunque la función en flatMap devuelve una lista de elementos, flatMap devuelve un RDD que tiene todos los elementos de la lista de manera plana (no una lista).

todos los ejemplos son buenos ... Aquí hay una buena ilustración visual ... cortesía de la fuente: DataFlair training of spark

Mapa: un mapa es una operación de transformación en Apache Spark. Se aplica a cada elemento de RDD y devuelve el resultado como un nuevo RDD. En el Mapa, el desarrollador de operaciones puede definir su propia lógica empresarial personalizada. La misma lógica se aplicará a todos los elementos de RDD.

La mapfunción Spark RDD toma un elemento como proceso de entrada de acuerdo con el código personalizado (especificado por el desarrollador) y devuelve un elemento a la vez. Map transforma un RDD de longitud N en otro RDD de longitud N. Los RDD de entrada y salida normalmente tendrán el mismo número de registros.

Ejemplo de mapuso de 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))

Mapa plano :

A flatMapes una operación de transformación. Se aplica a cada elemento de RDD y devuelve el resultado como nuevo RDD. Es similar a Map, pero FlatMap permite devolver 0, 1 o más elementos de la función de mapa. En la operación FlatMap, un desarrollador puede definir su propia lógica empresarial personalizada. La misma lógica se aplicará a todos los elementos del RDD.

¿Qué significa "aplanar los resultados"?

Una función FlatMap toma un elemento como proceso de entrada de acuerdo con el código personalizado (especificado por el desarrollador) y devuelve 0 o más elementos a la vez. flatMap() transforma un RDD de longitud N en otro RDD de longitud M.

Ejemplo de flatMapuso de 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)

La diferencia se puede ver en el siguiente código de ejemplo de 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 y Map transforman la colección.

Diferencia:

map (func)
Devuelve un nuevo conjunto de datos distribuido formado al pasar cada elemento de la fuente a través de una función func.

flatMap (func)
Similar a map, pero cada elemento de entrada se puede asignar a 0 o más elementos de salida (por lo que func debería devolver una Seq en lugar de un solo elemento).

La función de transformación:
mapa : un elemento de entrada -> un elemento de salida.
flatMap : un elemento dentro -> 0 o más elementos fuera (una colección).

RDD.map devuelve todos los elementos en una sola matriz

RDD.flatMap devuelve elementos en matrices de matriz

supongamos que tenemos texto en el archivo text.txt como

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

Usando el mapa

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

salida:

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

Usando flatMap

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

salida:

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

Para todos aquellos que han querido PySpark relacionados:

Transformación de ejemplo: flatMap

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

['Hola Qué estás haciendo']

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

Rastreo (última llamada más reciente): Archivo "", línea 1, en AttributeError: el objeto 'list' no tiene atributo 'split'

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

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


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

[['hola', 'qué', 'son', 'tú', 'haciendo'], ['esto', 'es', 'rak']]

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

['hola', 'qué', 'son', 'tú', 'haciendo', 'esto', 'es', 'rak']

Espero eso ayude :)

map: Devuelve un nuevo RDDaplicando una función a cada elemento de RDD. La función en .map solo puede devolver un elemento.

flatMap: Similar a un mapa, devuelve una nueva RDDmediante la aplicación de una función a cada elemento de la RDD, pero la salida se aplana.

Además, function in flatMappuede devolver una lista de elementos (0 o más)

Por ejemplo:

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

Salida: [[1, 2], [1, 2, 3], [1, 2, 3, 4]]

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

Salida: el aviso o / p se aplana en una sola lista [1, 2, 1, 2, 3, 1, 2, 3, 4]

Fuente: https://www.linkedin.com/pulse/difference-between-map-flatmap-transformations-spark-pyspark-pandey/

mapa :

es un método de orden superior que toma una función como entrada y la aplica a cada elemento en el RDD de origen.

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

mapa plano:

Un método de orden superior y una operación de transformación que toma una función de entrada.

Diferencia en la salida de map y flatMap:

1)flatMap

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

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

Salida:

 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()

Salida:

3 6 6 3 8

• map (func) Devuelve un nuevo conjunto de datos distribuido formado al pasar cada elemento de la fuente a través de una función declarada func.so map () es un término único

ratos

• flatMap (func) Similar a map, pero cada elemento de entrada se puede asignar a 0 o más elementos de salida, por lo que func debería devolver una secuencia en lugar de un solo elemento.