the below code is working fine, but as this is new for me, please help to improve performance. as I have not included yet my complex logic but confused over collect and parallelize. Aim is to collect patients id and check there respective visiting data and run it over cluster in parallel to improve performance.I am querying data from hive.class ClinicalMeasures extends Serializable {  var logger = Logger.getLogger(this.getClass())  val utility = new DateUtil;  val sc = ContextUtil.getSparkContext()  val hc = ContextUtil.getHiveContext(sc)  /*   * @param1 : wait time, @param2 : patients details, @param3: visit details   * @return : execution time in string   * @Description : Will return all visit details of particular patient   */  def processIndividual(wait: Int, patient: org.apache.spark.sql.Row, visits: Array[org.apache.spark.sql.Row]): String = {    var out = new StringBuilder()    val processStart = utility.getTimeInMillis()    try {      for (x <- visits) {    out.append(,  + x.getAs(patientid) + : + x.getAs(visitid))      }      if (wait > 0) {    Thread.sleep(wait)      }      logger.info(single function executed successfully)      return Time:  + utility.getFormattedTime() + ,  + utility.getExecutionTime(processStart) +  millis + out.toString()    } catch {      case e: Exception =>    println(exception caught in single function:  + e);    return false    }  }  /*   * @param1 : Batch size, @param2 : wait time, @param3 : patients details, @param4: visit details   * @Description : Run individual process in batch    */  def processBatch(batch: Int, wait: Int, patients: Array[org.apache.spark.sql.Row], visits: Array[org.apache.spark.sql.Row]) = {    try {      val start = utility.getTimeInMillis()      //println(Processing batch @  + utility.getFormattedTime() +   + patients.mkString(,))      logger.info(Processing batch @  + utility.getFormattedTime() +   + patients.mkString(,))      val out = sc.parallelize(patients, batch).map(r => processIndividual(wait, r, visits.filter(f => f.getAs(patientid) == r.getAs(patientid)))).collect()      for (x <- out) {    //println(X vlaue ===== >  + x)    logger.info(X vlaue ===== >  + x)      }      println(Batch took  + utility.getExecutionTime(start) +  millis)    } catch {      case e: Exception => println(exception caught in processbatch function:  + e);    }  }  def getPatients(): Array[org.apache.spark.sql.Row] = {    return hc.sql(SELECT patientid FROM table1 order by patientid).collect()  }  def getVisitDetails(): org.apache.spark.sql.DataFrame = {    return hc.sql(SELECT patientid, visitid FROM table2)  }  /*   * @Description : Executes measures on all patients as per patient visits    * @param1 : Fetch size, @param2 : Batch size, @param3 : wait time   *      */  def processMeasures(fetch: Int, batch: Int, wait: Int) = {    try {      println(Processing started @  + utility.getFormattedTime())      val processStart = utility.getTimeInMillis()      val patients = getPatients()      val visit = getVisitDetails()      val count = patients.length      val fetches = if(count % fetch > 0) (count / fetch + 1) else (count / fetch)      println(sPatient count $count, fetches $fetches);      for (i <- 0 to fetches.toInt - 1) {    val startFetch = i * fetch    val endFetch = math.min((i + 1) * fetch, count.toInt) - 1    val fetchSize = endFetch - startFetch + 1    val fetchClause = patientid >=  + patients(startFetch).get(0) +  and patientid <=  + patients(endFetch).get(0)    println(sFetching from $startFetch to $endFetch, $fetchSize, clause $fetchClause);    val fetchVisit = visit.filter(fetchClause).collect()    val batches = if (fetchSize % batch > 0) (fetchSize / batch + 1) else (fetchSize / batch)    println(sBatches $batches);    for (j <- 0 to batches.toInt - 1) {      val startBatch = j * batch      val endBatch = math.min((j + 1) * batch, fetch.toInt) - 1      println(sBatch from $startBatch to $endBatch);      val batchVisits = fetchVisit.filter(g => g.getAs[Long](patientid) >= patients(i*fetch + startBatch).getLong(0) && g.getAs[Long](patientid) <= patients(math.min(i*fetch + endBatch + 1, endFetch)).getLong(0))      processBatch(batch, wait, patients.slice(i * fetch + startBatch, i * fetch + endBatch + 1), batchVisits)    }      }      println(Processing took  + utility.getExecutionTime(processStart) +  millis)    } catch {      case e: Exception => println(exception caught in processMeasures function:  + e);    }  }}
collect and parallelize in spark performance
performance;scala;apache spark;hiveql