[apache-spark] map과 flatMap의 차이점은 무엇이며 각각의 좋은 사용 사례는 무엇입니까?

spark-shell세션 으로서의 차이점의 예는 다음과 같습니다 .

먼저 일부 데이터-두 줄의 텍스트 :

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

rdd.collect

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

이제 map길이 N의 RDD를 길이 N의 다른 RDD로 변환합니다.

예를 들어 두 줄에서 두 줄 길이로 매핑됩니다.

rdd.map(_.length).collect

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

그러나 flatMap(느슨하게 말하면) 길이 N의 RDD를 N 개의 모음으로 변환 한 다음 단일 RDD 결과로 평평하게 만듭니다.

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

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

한 줄에 여러 단어가 있고 여러 줄이 있지만 단어의 단일 출력 배열로 끝납니다.

설명을 위해, 줄 모음에서 단어 모음으로의 flatMapping은 다음과 같습니다.

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

따라서 입력 및 출력 RDD의 크기는 일반적으로 서로 다릅니다 flatMap.

함수 map와 함께 사용하려고 시도하면 입력 당 정확히 하나의 결과가 있어야하기 때문에 split중첩 구조 (유형이있는 단어 배열의 RDD)로 끝났습니다 RDD[Array[String]].

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

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

마지막으로 유용한 특수 사례 중 하나는 답변을 반환하지 않는 함수를 사용하여 매핑하는 것이므로를 반환합니다 Option. 다음 flatMap을 반환하는 요소 None에서 값 을 반환 하고 추출하는 요소를 필터링하는 데 사용할 수 있습니다 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)

(여기서 Option은 하나의 요소 또는 0 개의 요소가있는 목록과 비슷하게 작동 함)

답변

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

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

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

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

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

myRDD.map(x => x*2)

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

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

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)

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

1
2
3
-
4
-
5

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

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