Spark Streaming производительность скала резко медленный - PullRequest
Новая
       18

Spark Streaming производительность скала резко медленный

0 голосов
/ 25 апреля 2018

У меня есть следующий код: - 

case class event(imei: String, date: String, gpsdt: String,dt: String,id: String)
case class historyevent(imei: String, date: String, gpsdt: String)
object kafkatesting {
def main(args: Array[String]) {

val clients = new RedisClientPool("192.168.0.40", 6379)
val conf = new SparkConf()
  .setAppName("KafkaReceiver")
  .set("spark.cassandra.connection.host", "192.168.0.40")
  .set("spark.cassandra.connection.keep_alive_ms", "20000")
  .set("spark.executor.memory", "3g")
  .set("spark.driver.memory", "4g")
  .set("spark.submit.deployMode", "cluster")
  .set("spark.executor.instances", "4")
  .set("spark.executor.cores", "3")
  .set("spark.streaming.backpressure.enabled", "true")
  .set("spark.streaming.backpressure.initialRate", "100")
  .set("spark.streaming.kafka.maxRatePerPartition", "7")

val sc = SparkContext.getOrCreate(conf)
val ssc = new StreamingContext(sc, Seconds(10))
val sqlContext = new SQLContext(sc)
val kafkaParams = Map[String, String](
  "bootstrap.servers" -> "192.168.0.113:9092",
  "group.id" -> "test-group-aditya",
  "auto.offset.reset" -> "largest")

val topics = Set("random")
val kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)

kafkaStream.foreachRDD { rdd =>

  val updatedRDD = rdd.map(a =>
    {
      implicit val formats = DefaultFormats
      val jValue = parse(a._2)
      val fleetrecord = jValue.extract[historyevent]
      val hash = fleetrecord.imei + fleetrecord.date + fleetrecord.gpsdt
      val md5Hash = DigestUtils.md5Hex(hash).toUpperCase()
      val now = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(Calendar.getInstance().getTime())

      event(fleetrecord.imei, fleetrecord.date, fleetrecord.gpsdt, now, md5Hash)
    })
    .collect()

  updatedRDD.foreach(f =>
    {
      clients.withClient {
        client =>
          {
            val value = f.imei + " , " + f.gpsdt
            val zscore = Calendar.getInstance().getTimeInMillis
            val key = new SimpleDateFormat("yyyy-MM-dd").format(Calendar.getInstance().getTime())
            val dt = new SimpleDateFormat("HH:mm:ss").format(Calendar.getInstance().getTime())
            val q1 = "00:00:00"
            val q2 = "06:00:00"
            val q3 = "12:00:00"
            val q4 = "18:00:00"
            val quater = if (dt > q1 && dt < q2) {
              System.out.println(dt + " lies in quarter 1");
              " -> 1"
            } else if (dt > q2 && dt < q3) {
              System.out.println(dt + " lies in quarter 2");
              " -> 2"
            } else if (dt > q3 && dt < q4) {
              System.out.println(dt + " lies in quarter 3");
              " -> 3"
            } else {
              System.out.println(dt + " lies in quarter 4");
              " -> 4"
            }
            client.zadd(key + quater, zscore, value)
            println(f.toString())
          }
      }
    })
  val collection = sc.parallelize(updatedRDD)
  collection.saveToCassandra("db", "table", SomeColumns("imei", "date", "gpsdt","dt","id"))
}

ssc.start()
ssc.awaitTermination()
}
}

Я использую этот код для вставки данных из Kafka в Cassandra и Redis, но сталкиваюсь со следующими проблемами: -

1) приложение создает длинную очередь активных пакетов, пока предыдущий пакет обрабатывается в данный момент. Итак, я хочу получить следующую партию только после завершения выполнения предыдущей партии.

2) У меня есть кластер из четырех узлов, который обрабатывает каждый пакет, но для выполнения 700 записей требуется около 30-40 секунд.

Мой код оптимизирован или мне нужно работать над своим кодом для повышения производительности?

1 Ответ

0 голосов
/ 25 апреля 2018

Да, вы можете делать все свои вещи внутри mapPartition. Существуют API-интерфейсы из datastax, которые позволяют напрямую сохранять Dstream. Вот как вы можете сделать это для C *. 

val partitionedDstream = kafkaStream.repartition(5) //change this value as per your data and spark cluster

//Now instead of iterating each RDD work on each partition.
val eventsStream: DStream[event] = partitionedDstream.mapPartitions(x => {
  val lst = scala.collection.mutable.ListBuffer[event]()
  while (x.hasNext) {
    val a = x.next()
    implicit val formats = DefaultFormats
    val jValue = parse(a._2)
    val fleetrecord = jValue.extract[historyevent]
    val hash = fleetrecord.imei + fleetrecord.date + fleetrecord.gpsdt
    val md5Hash = DigestUtils.md5Hex(hash).toUpperCase()
    val now = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(Calendar.getInstance().getTime())
    lst += event(fleetrecord.imei, fleetrecord.date, fleetrecord.gpsdt, now, md5Hash)
  }
  lst.toList.iterator
})

eventsStream.cache() //because you are using same Dstream for C* and Redis

//instead of collecting each RDD save whole Dstream at once
import com.datastax.spark.connector.streaming._
eventsStream.saveToCassandra("db", "table", SomeColumns("imei", "date", "gpsdt", "dt", "id"))

Также Кассандра принимает timestamp в качестве значения Long, так что вы также можете изменить некоторую часть своего кода, как показано ниже 

val now = System.currentTimeMillis()

//also change your case class to take `Long` instead of `String`
case class event(imei: String, date: String, gpsdt: String, dt: Long, id: String)

Точно так же вы можете изменить на Redis.

...