火花地图与 foreachRdd

Question

我们有一个 spark 流应用程序，我们从 kafka 接收 dstream 并需要存储到 dynamoDB ....我正在尝试两种方法来做到这一点，如下面的代码所述

requestsWithState 是一个 Dstream

带有 foreachRDD 的代码片段 1：

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)

代码片段 1 工作正常并填充数据库......第二个代码 sn-p 不起作用......我们是新手，很想知道它背后的原因和获取方法它工作吗？ ........我们正在试验的原因（我们知道这是一个转换，foreachRdd 是一个动作）对于我们的集群负载很重的用例来说 foreachRdd 非常慢，我们发现如果我们能让它工作，那张地图会快得多.....请帮助我们让地图代码工作

Answer 1

Map 是 Spark 中的一种转换（惰性转换），除非您在 this 之后调用 spark 操作，否则它不会执行。 对于 Spark 转换和操作，请参阅以下链接 http://spark.apache.org/docs/latest/programming-guide.html#transformations

Answer 2

DStream.map 返回另一个流。您应该在该流而不是原始流上调用 print。

所以在 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)

Answer 3

map 的版本没有任何动作，.map 不是动作，而是变换。

没有动作就不会执行转换。

参见例如http://training.databricks.com/visualapi.pdf 或 http://spark.apache.org/docs/latest/programming-guide.html#transformations