Qual a diferença entre um RDD map e um mapPartitionsmétodo? E se flatMapcomporta como mapou como mapPartitions? Obrigado.

(editar) ie qual é a diferença (semanticamente ou em termos de execução) entre

  def map[A, B](rdd: RDD[A], fn: (A => B))
               (implicit a: Manifest[A], b: Manifest[B]): RDD[B] = {
    rdd.mapPartitions({ iter: Iterator[A] => for (i <- iter) yield fn(i) },
      preservesPartitioning = true)
  }

E:

  def map[A, B](rdd: RDD[A], fn: (A => B))
               (implicit a: Manifest[A], b: Manifest[B]): RDD[B] = {
    rdd.map(fn)
  }

Qual é a diferença entre o mapa de um RDD e o método mapPartitions?

O mapa do método converte cada elemento do RDD de origem em um único elemento do RDD resultante, aplicando uma função. mapPartitions converte cada partição do RDD de origem em vários elementos do resultado (possivelmente nenhum).

E o flatMap se comporta como map ou como mapPartitions?

O flatMap também não funciona em um único elemento (as map) e produz vários elementos do resultado (as mapPartitions).

Criança levada. DICA :

Sempre que você tiver uma inicialização pesada que deve ser feita uma vez para muitos RDDelementos, e não uma vez por RDDelemento, e se essa inicialização, como a criação de objetos a partir de uma biblioteca de terceiros, não puder ser serializada (para que o Spark possa transmiti-lo pelo cluster para os nós do trabalhador), use em mapPartitions()vez de map(). mapPartitions()prevê que a inicialização seja feita uma vez por tarefa / thread / partição do trabalhador, em vez de uma vez por RDDelemento de dados, por exemplo: veja abaixo.

val newRd = myRdd.mapPartitions(partition => {
  val connection = new DbConnection /*creates a db connection per partition*/

  val newPartition = partition.map(record => {
    readMatchingFromDB(record, connection)
  }).toList // consumes the iterator, thus calls readMatchingFromDB 

  connection.close() // close dbconnection here
  newPartition.iterator // create a new iterator
})

Q2 se flatMapcomporta como mapa ou comomapPartitions ?

Sim. veja o exemplo 2 deflatmap .. é auto-explicativo.

Q1 Qual é a diferença entre um RDDmap e ummapPartitions

map trabalha a função que está sendo utilizada no nível por elemento enquanto mapPartitions exerce a função no nível da partição.

Cenário de exemplo : se tivermos 100 mil elementos em um determinadoRDD partição , dispararemos a função que está sendo usada pela transformação de mapeamento 100K vezes quando usamos map.

Por outro lado, se usarmos mapPartitions , chamaremos a função específica apenas uma vez, mas passaremos todos os 100 mil registros e receberemos de volta todas as respostas em uma chamada de função.

Haverá ganho de desempenho desde map funciona em uma função específica tantas vezes, especialmente se a função estiver fazendo algo caro a cada vez que não seria necessário se passássemos todos os elementos de uma só vez (no caso de mappartitions).

mapa

Aplica uma função de transformação em cada item do RDD e retorna o resultado como um novo RDD.

Variantes de listagem

mapa de def [U: ClassTag] (f: T => U): RDD [U]

Exemplo:

val a = sc.parallelize(List("dog", "salmon", "salmon", "rat", "elephant"), 3)
 val b = a.map(_.length)
 val c = a.zip(b)
 c.collect
 res0: Array[(String, Int)] = Array((dog,3), (salmon,6), (salmon,6), (rat,3), (elephant,8)) 

mapPartitions

Este é um mapa especializado chamado apenas uma vez para cada partição. Todo o conteúdo das respectivas partições está disponível como um fluxo sequencial de valores por meio do argumento de entrada (Iterarator [T]). A função personalizada deve retornar outro iterador [U]. Os iteradores de resultados combinados são convertidos automaticamente em um novo RDD. Observe que as tuplas (3,4) e (6,7) estão ausentes no resultado a seguir devido ao particionamento escolhido.

preservesPartitioningindica se a função de entrada preserva o particionador, que deve ser, a falsemenos que seja um par RDD e a função de entrada não modifique as chaves.

Variantes de listagem

def mapPartitions [U: ClassTag] (f: Iterador [T] => Iterador [U], preservesPartitioning: Boolean = false): RDD [U]

Exemplo 1

val a = sc.parallelize(1 to 9, 3)
 def myfunc[T](iter: Iterator[T]) : Iterator[(T, T)] = {
   var res = List[(T, T)]()
   var pre = iter.next
   while (iter.hasNext)
   {
     val cur = iter.next;
     res .::= (pre, cur)
     pre = cur;
   }
   res.iterator
 }
 a.mapPartitions(myfunc).collect
 res0: Array[(Int, Int)] = Array((2,3), (1,2), (5,6), (4,5), (8,9), (7,8)) 

Exemplo 2

val x = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9,10), 3)
 def myfunc(iter: Iterator[Int]) : Iterator[Int] = {
   var res = List[Int]()
   while (iter.hasNext) {
     val cur = iter.next;
     res = res ::: List.fill(scala.util.Random.nextInt(10))(cur)
   }
   res.iterator
 }
 x.mapPartitions(myfunc).collect
 // some of the number are not outputted at all. This is because the random number generated for it is zero.
 res8: Array[Int] = Array(1, 2, 2, 2, 2, 3, 3, 3, 3, 3, 3, 3, 3, 3, 4, 4, 4, 4, 4, 4, 4, 5, 7, 7, 7, 9, 9, 10) 

O programa acima também pode ser escrito usando o flatMap da seguinte maneira.

Exemplo 2 usando flatmap

val x  = sc.parallelize(1 to 10, 3)
 x.flatMap(List.fill(scala.util.Random.nextInt(10))(_)).collect

 res1: Array[Int] = Array(1, 2, 3, 3, 3, 4, 4, 4, 4, 4, 4, 4, 4, 4, 5, 5, 6, 6, 6, 6, 6, 6, 6, 6, 7, 7, 7, 8, 8, 8, 8, 8, 8, 8, 8, 9, 9, 9, 9, 9, 10, 10, 10, 10, 10, 10, 10, 10) 

Conclusão:

mapPartitionsa transformação é mais rápida do que mapuma vez que chama sua função uma vez / partição, não uma vez / elemento.

Leitura adicional: foreach Vs foreachPartitions Quando usar o quê?

Mapa :

1. Ele processa uma linha por vez, muito semelhante ao método map () do MapReduce.
2. Você retorna da transformação após cada linha.

MapPartitions

1. Ele processa a partição completa de uma só vez.
2. Você pode retornar da função apenas uma vez após o processamento de toda a partição.
3. Todos os resultados intermediários precisam ser mantidos na memória até que você processe toda a partição.
4. Fornece as funções setup () map () e cleanup () do MapReduce

Map Vs mapPartitions http://bytepadding.com/big-data/spark/spark-map-vs-mappartitions/

Spark Map http://bytepadding.com/big-data/spark/spark-map/

Spark mapPartitions http://bytepadding.com/big-data/spark/spark-mappartitions/