Spark Streaming - Dataframe Collect Performance Issue

Question

I m trying to improve spark streaming application for better performance. In each streaming cycle, i m generating a new dataframe for each record consuming from topic and i need to collect value list from this dataframe, to use in the analytic model stage.

Here is my application steps :

1- Read from kafka
For Loop
    2- Generate a new dataframe by joining static dataframe with new topic dataframe (Columns : key,value)
    3- Collect value list from dataframe. (CollectDF function)
    4- Calling pmml model
    ...
    2- Generate a new dataframe by joining static dataframe with new topic dataframe (Columns : key,value)
    3- Collect value list from dataframe. (CollectDF function)
    4- Calling pmml model
    ...
    
    If there are 10 record in topic, this cycle is runing 10 times. At first, CollectDF process takes 1-2 seconds but after a few cycle in the loop, this process takes 8-10 seconds.
    Actually i dont understand how this is possible. How can i keep the process time stable ?
     

      kafkaStream.foreachRDD(rdd => {
        stream_df.collect().foreach { row =>
        ...
        val model_feature_list = CollectDF(df_model)
        val predictions = model.predict(model_feature_list)
        }
    }
    
      def CollectDF(df_modelparam : DataFrame): Array[Int] ={
        val x : Map[String, Int] = df_modelparam.collect.map( r => {
          val key = r(0).toString
          val value = r(1).toString.toInt
          (key -> value)
        }
        ).toMap.toSortedMap
        var x_arr = x.values.toArray
        x_arr
      }

Thanks in advance

Shane · Answer

May I know the reason for collecting the data to driver ?

Ideally you should try to avoid collect() function in spark streaming usecases as its a costly operation and might slow things down.

May be you can try something like below on streaming Dataframe itself instead of collecting the data to driver.

streamingDF.mapPartitions(rowIterator=>{
rowIterator.foreach(row =>{
          val key = row(0).toString
          val value = row(1).toString.toInt
          (key -> value)
          // analytical use case on the above key, value being created
   }
}

Spark Streaming - Dataframe Collect Performance Issue

1 Answers