Спарк карта против foreachRdd

Question

У нас есть искра потокового приложения, где мы получаем в dstream от Кафки и нужно хранить в dynamoDB .... я буду экспериментировать с двумя способами, чтобы сделать это, как описано в коде ниже

requestsWithState является Dstream

Фрагмент кода 1 с foreachRDD:

requestsWithState.foreachRDD { rdd => 
    println("Data being populated to Pulsar") 
    rdd.foreach { case (id, eventStream) => 
    println("id is " + id + " Event is " + eventStream) 
    DBUtils.putItem(dynamoConnection, id, eventStream.toString()) 
    } 
} 

фрагмент кода 2 с картой:

requestsWithState.map (rdd => { rdd match { 
     case (id, eventStream) => { 
      println("id is " + id + " Event is " + eventStream) 
      val dynamoConnection = setupDynamoClientConnection() 
      DBUtils.putItem(dynamoConnection, id, eventStream.toString()) 
     } 
     } 
    }) 

requestsWithState.print(1) 

Код Snippet1 работа в порядке и заполнит базу данных ... второй фрагмент кода не работает .... мы новичок, чтобы зажечь и хотел бы знать причину этого и путь к заставить его работать? ........ причина, по которой мы экспериментируем (мы знаем, что это преобразование, и foreachRdd - это действие) foreachRdd очень медленный для нашего случая использования с большой нагрузкой на кластер, и мы обнаружили, что карта намного быстрее, если мы можем заставить его работать ..... пожалуйста, помогите нам получить код карты, работающий

Answer 1

Карта - это тип преобразования (ленивая трансформация) в Spark и не будет выполняться, если после этого вы не вызовете искрение. Для трансформации и действий Спарк, относится к ссылке ниже http://spark.apache.org/docs/latest/programming-guide.html#transformations

Answer 2

Вариант с map не имеет каких-либо действий и .map не действие, но преобразование.

Преобразования не выполняются без каких-либо действий.

См., Например, http://training.databricks.com/visualapi.pdf или http://spark.apache.org/docs/latest/programming-guide.html#transformations

Answer 3

DStream.map возвращает другой поток. Вы должны вызывать печать в этом потоке, а не в оригинале.

Таким образом, в Scala:

val transformedStream = requestsWithState.map (rdd => { rdd match { 
     case (id, eventStream) => { 
      println("id is " + id + " Event is " + eventStream) 
      val dynamoConnection = setupDynamoClientConnection() 
      DBUtils.putItem(dynamoConnection, id, eventStream.toString()) 
     } 
     } 
    }) 

transformedStream.print(1)