Spark : DB connection per Spark RDD partition and do mapPartition

Question

I want to do a mapPartitions on my 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
      })

But, this gives me a connection already closed exception, as expected because before the control reaches the .map() my connection is closed. I want to create a connection per RDD partition, and close it properly. How can I achieve this?

Thanks!

I've never tried this but I believe you ought creating your connections outside the closure and thus close after it's finished. — eliasah, Jun 17 '16 at 12:16
That would lead to a single connection. I want one connection per partition — void, Jun 17 '16 at 12:17
I'm aware that it leads to one connection. I believe that you should try the try catch regular approach for DBConnection — eliasah, Jun 17 '16 at 12:18
which database you are using? cant it be solved with connection pooling(opening the connection with minimum number of connections (may be =number of partitions) , reusing the connection when its closed)? where when you close the connection then its not actual close of the connection instead it will return to the pool. some data bases you can also check whether connection is closed or not through its api — Ram Ghadiyaram, Jun 17 '16 at 13:10

score 10 · Accepted Answer · edited May 23 '17 at 11:33

10

As mentioned in the discussion here - the issue stems from the laziness of map operation on the iterator partition. This laziness means that for each partition, a connection is created and closed, and only later (when RDD is acted upon), readMatchingFromDB is called.

To resolve this, you should force an eager traversal of the iterator before closing the connection, e.g. by converting it into a list (and then back):

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
})

edited May 23 '17 at 11:33

Community

1
1

answered Jun 20 '16 at 05:29

Tzach Zohar

37,442
3
79
85

Can you provide me with an actual example of the above settjng ghe conndctikn, etc. pls? Or point somewhere to? thx – thebluephantom May 01 '18 at 06:20
setting the connection, pardon the spelling on mobile – thebluephantom May 01 '18 at 22:01
What if we need to handle exception when creating the db connection? – Feiyu Zhou Jul 07 '22 at 05:04
@FeiyuZhou what prevents you from doing that? You can surround `new DbConnection` with a try-catch and handle such exceptions as you like. You just need to decide what to do in such a case (return empty partition, throw the exception etc.) – Tzach Zohar Jul 07 '22 at 12:49

score 0 · Answer 2 · answered Dec 06 '18 at 20:17

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
}

}

Spark : DB connection per Spark RDD partition and do mapPartition

2 Answers2

Linked