1
\$\begingroup\$

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(s"Patient 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(s"Fetching 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(s"Batches $batches");

    for (j <- 0 to batches.toInt - 1) {
      val startBatch = j * batch
      val endBatch = math.min((j + 1) * batch, fetch.toInt) - 1

      println(s"Batch 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);
    }
  }
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.