Spark：每个Spark RDD分区和do mapPartition的DB连接

Question

我想在我的spark rdd上做一个映射，

    val newRd = myRdd.mapPartitions(
      partition => {

        val connection = new DbConnection /*creates a db connection per partition*/

        val newPartition = partition.map(
           record => {
             readMatchingFromDB(record, connection)
         })
        connection.close()
        newPartition
      })

但是，这给了我一个已经关闭的连接异常，正如预期的那样，因为在控件到达. map（）之前，我的连接是关闭的。我想为每个RDD分区创建一个连接，并正确地关闭它。我如何实现这一点？

谢谢

Answer 1

rdd.foreachPartitionAsync(iterator->{

// this object will be cached inside each executor JVM. For the first time, the //connection will be created and hence forward, it will be reused. 
// Very useful for streaming apps
DBConn conn=DBConn.getConnection()
while(iterator.hasNext()) {
  conn.read();
}

});

public class DBConn{
private static dbObj=null;

//Create a singleton method that returns only one instance of this object
}

}

Answer 2

正如在这里的讨论中提到的，这个问题源于迭代器分区上映射操作的惰性。这种惰性意味着，对于每个分区，都会创建并关闭一个连接，只有在以后（对RDD进行操作时）才会调用readMatchingFromDB。

要解决这个问题，您应该在关闭连接之前强制遍历迭代器，例如将其转换为列表（然后返回）：

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()
  newPartition.iterator // create a new iterator
})