我们有一个spark流应用程序（下面是代码），它从kafka获取数据，并在将数据插入mongodb之前（对每条消息）进行一些转换。我们有一个中间件应用程序，它将消息（批量）推入kafka，并等待spark流应用程序的确认（对于每条消息）。如果在将消息发送到kafka后的特定时间段（5秒）内中间件没有收到确认，则中间件应用程序重新发送消息。spark流应用程序能够接收大约50-100条消息（一批），并在5秒内发送所有消息的确认。但是，如果中间件应用程序推送超过100条消息，则会导致中间件应用程序由于spark streaming发送确认的延迟而重新发送消息。在我们当前的实现中，我们每次要发送确认时都会创建producer，这需要3-4秒。

package com.testing

import org.apache.spark.streaming._
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.{ Seconds, StreamingContext }
import org.apache.spark.{ SparkConf, SparkContext }
import org.apache.spark.streaming.kafka._
import org.apache.spark.sql.{ SQLContext, Row, Column, DataFrame }
import java.util.HashMap
import org.apache.kafka.clients.producer.{ KafkaProducer, ProducerConfig, ProducerRecord }
import scala.collection.mutable.ArrayBuffer
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types._

import org.joda.time._
import org.joda.time.format._

import org.json4s._
import org.json4s.JsonDSL._
import org.json4s.jackson.JsonMethods._
import com.mongodb.util.JSON

import scala.io.Source._
import java.util.Properties
import java.util.Calendar

import scala.collection.immutable
import org.json4s.DefaultFormats

object Sample_Streaming {

  def main(args: Array[String]) {

    val sparkConf = new SparkConf().setAppName("Sample_Streaming")
      .setMaster("local[4]")

    val sc = new SparkContext(sparkConf)
    sc.setLogLevel("ERROR")

    val sqlContext = new SQLContext(sc)
    val ssc = new StreamingContext(sc, Seconds(1))

    val props = new HashMap[String, Object]()

    val bootstrap_server_config = "127.0.0.100:9092"
    val zkQuorum = "127.0.0.101:2181"

    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap_server_config)
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")

    val TopicMap = Map("sampleTopic" -> 1)
    val KafkaDstream = KafkaUtils.createStream(ssc, zkQuorum, "group", TopicMap).map(_._2)

      val schemaDf = sqlContext.read.format("com.mongodb.spark.sql.DefaultSource")
        .option("spark.mongodb.input.uri", "connectionURI")
        .option("spark.mongodb.input.collection", "schemaCollectionName")
        .load()

      val outSchema = schemaDf.schema
      var outDf = sqlContext.createDataFrame(sc.emptyRDD[Row], outSchema)

    KafkaDstream.foreachRDD(rdd => rdd.collect().map { x =>
      {
        val jsonInput: JValue = parse(x)

        /*Do all the transformations using Json libraries*/

        val json4s_transformed = "transformed json"

        val rdd = sc.parallelize(compact(render(json4s_transformed)) :: Nil)
        val df = sqlContext.read.schema(outSchema).json(rdd)

        df.write.option("spark.mongodb.output.uri", "connectionURI")
                  .option("collection", "Collection")
                  .mode("append").format("com.mongodb.spark.sql").save()

        val producer = new KafkaProducer[String, String](props)
        val message = new ProducerRecord[String, String]("topic_name", null, "message_received")

        producer.send(message)
        producer.close()

      }

    }

    )

    // Run the streaming job
    ssc.start()
    ssc.awaitTermination()
  }

}

因此，我们尝试了另一种方法，在foreachrdd之外创建producer，并在整个批处理间隔内重用它（下面是代码）。这似乎有帮助，因为我们不是每次都要创建生产者发送确认。但由于某些原因，当我们在spark ui上监视应用程序时，流应用程序的内存消耗正在稳步增加，这与以前不同。我们尝试在spark submit中使用--num executors 1选项来限制由yarn启动的执行器的数量。

object Sample_Streaming {

    def main(args: Array[String]) {

    val sparkConf = new SparkConf().setAppName("Sample_Streaming")
      .setMaster("local[4]")

    val sc = new SparkContext(sparkConf)
    sc.setLogLevel("ERROR")

    val sqlContext = new SQLContext(sc)
    val ssc = new StreamingContext(sc, Seconds(1))

    val props = new HashMap[String, Object]()

    val bootstrap_server_config = "127.0.0.100:9092"
    val zkQuorum = "127.0.0.101:2181"

    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap_server_config)
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")

    val TopicMap = Map("sampleTopic" -> 1)
    val KafkaDstream = KafkaUtils.createStream(ssc, zkQuorum, "group", TopicMap).map(_._2)

      val schemaDf = sqlContext.read.format("com.mongodb.spark.sql.DefaultSource")
        .option("spark.mongodb.input.uri", "connectionURI")
        .option("spark.mongodb.input.collection", "schemaCollectionName")
        .load()

      val outSchema = schemaDf.schema
    val producer = new KafkaProducer[String, String](props)
    KafkaDstream.foreachRDD(rdd => 
          {

            rdd.collect().map ( x =>
            {

              val jsonInput: JValue = parse(x)

              /*Do all the transformations using Json libraries*/

              val json4s_transformed = "transformed json"

              val rdd = sc.parallelize(compact(render(json4s_transformed)) :: Nil)
              val df = sqlContext.read.schema(outSchema).json(rdd)

              df.write.option("spark.mongodb.output.uri", "connectionURI")
                        .option("collection", "Collection")
                        .mode("append").format("com.mongodb.spark.sql").save()

              val message = new ProducerRecord[String, String]("topic_name", null, "message_received")

              producer.send(message)
              producer.close()

            }

            )
        }

    )

    // Run the streaming job
    ssc.start()
    ssc.awaitTermination()
  }

}

我的问题是：
如何监控spark应用程序的内存消耗，目前我们每5分钟手动监控一次应用程序，直到它耗尽集群中的可用内存（每个2节点16gb）？
在使用spark streaming和kafka时，业界遵循的最佳实践是什么？