How to persist the list which we made dynamically from dataFrame in scala spark

user_13542005 0
Viewed 332
 def getAnimalName(dataFrame: DataFrame): List[String] = {
    dataFrame.select("animal").
      filter(col("animal").isNotNull && col("animal").notEqual("")).
      rdd.map(r => r.getString(0)).distinct().collect.toList
  }

I am basicaly Calling this function 2 times For getting the list for different purposes . I just want to know is there a way to retain the list in memory and we dont have to call the same function again and again to generate the list and only have to generate the list only one time in scala spark.

apache-spark scala
2 Answers

Try something as below and you can also check the performance using time func. Also find the code explanation inline

import org.apache.spark.rdd
import org.apache.spark.sql.functions._
import org.apache.spark.sql.{DataFrame, functions}

object HandleCachedDF {

  var cachedAnimalDF : rdd.RDD[String] = _
  def main(args: Array[String]): Unit = {
    val spark = Constant.getSparkSess
    val df = spark.read.json("src/main/resources/hugeTest.json") // Load your Dataframe
    val df1 = time[rdd.RDD[String]] {
      getAnimalName(df)
    }
    val resultList = df1.collect().toList
    val df2 = time{
      getAnimalName(df)
    }
    val resultList1 = df2.collect().toList
    println(resultList.equals(resultList1))
  }

  def getAnimalName(dataFrame: DataFrame): rdd.RDD[String] = {
    if (cachedAnimalDF == null) { // Check if this the first initialization of your dataframe
      cachedAnimalDF = dataFrame.select("animal").
        filter(functions.col("animal").isNotNull && col("animal").notEqual("")).
        rdd.map(r => r.getString(0)).distinct().cache() // Cache your dataframe
    }
    cachedAnimalDF // Return your cached dataframe
  }

  def time[R](block: => R): R = { // COmpute the time taken by function to execute
    val t0 = System.nanoTime()
    val result = block // call-by-name
    val t1 = System.nanoTime()
    println("Elapsed time: " + (t1 - t0) + "ns")
    result
  }

}
user_63035790

You would have to persist or cache at this point

dataFrame.select("animal").
      filter(col("animal").isNotNull && col("animal").notEqual("")).
      rdd.map(r => r.getString(0)).distinct().persist

and then call the function as follow

def getAnimalName(dataFrame: DataFrame): List[String] = {
    dataFrame.collect.toList
  }

as many times as you need it without repeat the process. I hope it helps.

user_85714980
Related