How to batch columns of spark dataframe, process with REST API and add it back?-CodePudding

I have a dataframe in spark and I need to process a particular column in that dataframe using a REST API. The API does some transformation to a string and returns a result string. The API can process multiple strings at a time.

I can iterate over the columns of the dataframe, collect n values of the column in a batch and call the api and then add it back to the dataframe, and continue with the next batch. But this seems like the normal way of doing it without taking advantage of spark.

Is there a better way to do this which can take advantage of spark sql optimiser and spark parallel processing?

CodePudding user response：

For Spark parallel processing you can use mapPartitions

case class Input(col: String)
case class Output ( col : String,new_col : String )
val data = spark.read.csv("/a/b/c").as[Input].repartiton(n)

 def declare(partitions: Iterator[Input]): Iterator[Output] ={
      val url = ""
      implicit val formats: DefaultFormats.type = DefaultFormats
      var list = new ListBuffer[Output]()
      val httpClient = 
      try {
        while (partitions.hasNext) {

          val x = partitions.next()
          val col = x.col

          val concat_url =""
          val apiResp = HttpClientAcceptSelfSignedCertificate.call(httpClient, concat_url)
          if (apiResp.isDefined) {
            val json = parse(apiResp.get)
          val new_col = (json \\"value_to_take_from_api").children.head.values.toString
            val output = Output(col,new_col)
          list =output
  
          }
          else {
            val new_col = "Not Found"
            val output = Output(col,new_col)
            list =output

          }
        }
      } catch {
 
        case e: Exception => println("api Exception with : "   e.getMessage)
      }
      finally {
        HttpClientAcceptSelfSignedCertificate.close(httpClient)
      }
      list.iterator
    }

val dd:Dataset[Output] =data.mapPartitions(x=>declare(x))