我不知道我是否正确理解spark如何处理数据库连接，以及如何在spark内部使用大量的数据库更新操作而不影响spark的工作。这是我一直在使用的代码片段（为了便于说明）：

val driver = new MongoDriver
val hostList: List[String] = conf.getString("mongo.hosts").split(",").toList
val connection = driver.connection(hostList)
val mongodb = connection(conf.getString("mongo.db"))
val dailyInventoryCol = mongodb[BSONCollection](conf.getString("mongo.collections.dailyInventory"))

val stream: InputDStream[(String,String)] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder, (String, String)](
  ssc, kafkaParams, fromOffsets,
  (mmd: MessageAndMetadata[String, String]) => (mmd.topic, mmd.message()));

def processRDD(rddElem: RDD[(String, String)]): Unit = {
    val df = rdd.map(line => {
        ...
    }).flatMap(x => x).toDF()

    if (!isEmptyDF(df)) {
        var mongoF: Seq[Future[dailyInventoryCol.BatchCommands.FindAndModifyCommand.FindAndModifyResult]] = Seq();

        val dfF2 = df.groupBy($"CountryCode", $"Width", $"Height", $"RequestType", $"Timestamp").agg(sum($"Frequency")).collect().map(row => {
        val countryCode = row.getString(0); val width = row.getInt(1); val height = row.getInt(2);
        val requestType = row.getInt(3); val timestamp = row.getLong(4); val frequency = row.getLong(5);
        val endTimestamp = timestamp + 24*60*60; //next day

        val updateOp = dailyInventoryCol.updateModifier(BSONDocument("$inc" -> BSONDocument("totalFrequency" -> frequency)), false, true)

        val f: Future[dailyInventoryCol.BatchCommands.FindAndModifyCommand.FindAndModifyResult] =
        dailyInventoryCol.findAndModify(BSONDocument("width" -> width, "height" -> height, "country_code" -> countryCode, "request_type" -> requestType,
         "startTs" -> timestamp, "endTs" -> endTimestamp), updateOp) 

        f
   })

   mongoF = mongoF ++ dfF2

   //split into small chunk to avoid drying out the mongodb connection
   val futureList: List[Seq[Future[dailyInventoryCol.BatchCommands.FindAndModifyCommand.FindAndModifyResult]]] = mongoF.grouped(200).toList

   //future list
   futureList.foreach(seqF => {
     Await.result(Future.sequence(seqF), 40.seconds)
   });     
}

stream.foreachRDD(processRDD(_))

基本上，我使用的是reactive mongo（scala），对于每个rdd，我将其转换为dataframe，分组/提取必要的数据，然后对mongo发起大量数据库更新查询。我想问：
我正在使用mesos在3台服务器上部署spark，并为mongo数据库多了一台服务器。这是处理数据库连接的正确方法吗。我关心的是，在spark作业开始时是否打开了数据库连接/轮询，并在spark的整个持续时间（周、月……）内正确地维护了数据库连接/轮询（尽管超时/网络错误故障转移），以及在每个批处理完成时是否关闭数据库连接/轮询？考虑到作业可能被安排在不同的服务器上？这是否意味着每批，它将打开不同的数据库连接集？
如果执行查询时发生异常会发生什么情况。该批的spark作业将失败？但下一批会继续吗？
如果有太多的查询（2000->+）在mongo数据库上运行update，并且执行时间超过配置的spark batch duration（2分钟），是否会导致问题？我注意到，在我当前的设置中，在abt 2-3天后，所有批处理都在spark webui上以“进程”的形式排队（如果我禁用mongo update部分，那么我可以运行一周而没有prob），没有一个批处理能够正确退出。它基本上会挂起所有批处理作业，直到我重新启动/重新提交作业。
谢谢。如果你能帮我解决这个问题，我将不胜感激。