在apache spark streaming中使用foreachRDD内部的db连接

Question

在spark streaming中，我希望在处理每个批处理之前查询db，将结果存储在一个可以序列化并通过网络发送给执行者的hashmap中。

class ExecutingClass implements Serializable {
 init(DB db) {

   try(JavaStreamingContext jsc = new JavaStreamingContext(...)) {

   JavaPairInputDStream<String,String> kafkaStream = getKafkaStream(jsc);

   kafkaStream.foreachRDD(rdd -> {
   // this part is supposed to execute in the driver
  Map<String, String> indexMap = db.getIndexMap();// connects to a db, queries the results as a map

  JavaRDD<String> results = processRDD(rdd, indexMap);

  ...  

 }


  }
    JavaRDD<String> processRDD(JavaPairRDD<String, String> rdd,       Map<String,String> indexMap) {
 ... 
    }
    }

在上面的代码中，indexMap应该在驱动程序中初始化，得到的map用于处理RDD。在foreachRDD闭包外部声明indexMap时没有问题，但在内部声明时会出现序列化错误。这是什么原因呢？

我之所以要这样做，是为了确保每个批处理都有数据库中的最新值。我怀疑这是由于foreachRDD的闭包试图序列化闭包之外的一切。

Answer 1

您正在使用forEachRdd中的db对象（它是db的实例），因此spark尝试序列化db，为了避免这种情况，我们需要在forEachRdd中创建db连接（或者），您可以使用对象池，如以下文章http://www.michael-noll.com/blog/2014/10/01/kafka-spark-streaming-integration-example-tutorial/中所讨论的