SB-27408 | Assessment Archival JOB - Deletion function added

data-products/src/main/scala/org/sunbird/analytics/archival/AssessmentArchivalJob.scala

@@ -0,0 +1,194 @@
+package org.sunbird.analytics.archival
+
+import com.datastax.spark.connector.{SomeColumns, toRDDFunctions}
+import org.apache.spark.sql.functions.{col, concat, lit, to_json, to_timestamp, weekofyear, year}
+import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.{DataFrame, SQLContext, SparkSession}
+import org.ekstep.analytics.framework.conf.AppConf
+import org.ekstep.analytics.framework.util.{JSONUtils, JobLogger, RestUtil}
+import org.ekstep.analytics.framework.{FrameworkContext, JobConfig, Level}
+import org.sunbird.analytics.archival.util.ArchivalRequest
+
+import java.util.concurrent.atomic.AtomicInteger
+import org.apache.spark.sql.functions._
+import org.sunbird.analytics.exhaust.util.ExhaustUtil
+import org.sunbird.analytics.util.Constants
+
+import scala.collection.immutable.List
+
+case class CollectionDetails(result: Result)
+case class Result(content: List[CollectionInfo])
+case class CollectionInfo(identifier: String)
+
+object AssessmentArchivalJob extends optional.Application with BaseArchivalJob {
+
+  case class Period(year: Int, weekOfYear: Int)
+  case class BatchPartition(collectionId: String, batchId: String, period: Period)
+  case class ArchivalMetrics(batch: BatchPartition,
+                             totalArchivedRecords: Option[Long],
+                             pendingWeeksOfYears: Option[Long],
+                             totalDeletedRecords: Option[Long]
+                            )
+
+  private val partitionCols = List("course_id", "batch_id", "year", "week_of_year")
+  private val columnWithOrder = List("course_id", "batch_id", "user_id", "content_id", "attempt_id", "created_on", "grand_total", "last_attempted_on", "total_max_score", "total_score", "updated_on", "question")
+
+  override def getClassName = "org.sunbird.analytics.archival.AssessmentArchivalJob"
+  override def jobName = "AssessmentArchivalJob"
+  override def jobId: String = "assessment-archival"
+  override def getReportPath = "assessment-archival/"
+  override def getReportKey = "assessment"
+  override def dateColumn = "updated_on"
+
+  override def archivalFormat(batch: Map[String,AnyRef]): String = {
+    val formatDetails = JSONUtils.deserialize[BatchPartition](JSONUtils.serialize(batch))
+    s"${formatDetails.batchId}_${formatDetails.collectionId}/${formatDetails.period.year}-${formatDetails.period.weekOfYear}"
+  }
+
+  override def dataFilter(requests: Array[ArchivalRequest], dataDF: DataFrame): DataFrame = {
+    var filteredDF = dataDF
+    for (request <- requests) {
+      if (request.archival_status.equals("SUCCESS")) {
+        val request_data = JSONUtils.deserialize[Map[String, AnyRef]](request.request_data)
+        filteredDF = dataDF.filter(
+          col("batch_id").equalTo(request.batch_id) &&
+          concat(col("year"), lit("-"), col("week_of_year")) =!= lit(request_data("year") + "-" + request_data("week"))
+        )
+      }
+    }
+    filteredDF
+  }
+
+  override def archiveData(requestConfig: Request, requests: Array[ArchivalRequest])(implicit spark: SparkSession, fc: FrameworkContext, config: JobConfig): List[ArchivalRequest] = {
+    try {
+      val archivalKeyspace = requestConfig.keyspace.getOrElse(AppConf.getConfig("sunbird.courses.keyspace"))
+      val date: String  = requestConfig.date.getOrElse(null)
+
+      var data = loadData(Map("table" -> requestConfig.archivalTable, "keyspace" -> archivalKeyspace, "cluster" -> "LMSCluster"), cassandraUrl, new StructType())
+      data = validateBatches(data, requestConfig)
+
+      val dataDF = generatePeriodInData(data)
+      val filteredDF = dataFilter(requests, dataDF)
+
+      val archiveBatchList = filteredDF.groupBy(partitionCols.head, partitionCols.tail: _*).count().collect()
+      val batchesToArchive: Map[String, Array[BatchPartition]] = archiveBatchList.map(f => BatchPartition(f.get(0).asInstanceOf[String], f.get(1).asInstanceOf[String], Period(f.get(2).asInstanceOf[Int], f.get(3).asInstanceOf[Int]))).groupBy(_.batchId)
+
+      archiveBatches(batchesToArchive, filteredDF, requestConfig)
+    } catch {
+      case ex: Exception =>
+        ex.printStackTrace()
+        JobLogger.log("archiveData: Exception with error message = " + ex.getMessage, None, Level.ERROR)
+        List()
+    }
+  }
+
+  def validateBatches(data: DataFrame, requestConfig: Request)(implicit spark: SparkSession, fc: FrameworkContext): DataFrame ={
+    implicit val sqlContext = new SQLContext(spark.sparkContext)
+    import sqlContext.implicits._
+
+    val filteredDF = if(requestConfig.batchId.isDefined && requestConfig.collectionId.isDefined) {
+      data.filter(col("batch_id") === requestConfig.batchId.get && col("course_id") === requestConfig.collectionId.get).persist()
+    } else if (requestConfig.batchFilters.isDefined) {
+      val batch = requestConfig.batchFilters.get.toDF("batch_id")
+      data.join(batch, Seq("batch_id"), "inner")
+    } else {
+      JobLogger.log("Neither batchId nor batchFilters present", None, Level.INFO)
+      data
+    }
+    if (requestConfig.query.isDefined) {
+      val res = searchContent(requestConfig.query.get)
+      filteredDF.join(res, col("course_id") === col("identifier"), "inner")
+    } else filteredDF
+
+  }
+
+  def searchContent(searchFilter: Map[String, AnyRef])(implicit spark: SparkSession, fc: FrameworkContext): DataFrame = {
+    // TODO: Handle limit and do a recursive search call
+    val apiURL = Constants.COMPOSITE_SEARCH_URL
+    val request = JSONUtils.serialize(searchFilter)
+    val response = RestUtil.post[CollectionDetails](apiURL, request).result.content
+    spark.createDataFrame(response).select("identifier")
+  }
+
+  def archiveBatches(batchesToArchive: Map[String, Array[BatchPartition]], data: DataFrame, requestConfig: Request)(implicit config: JobConfig): List[ArchivalRequest] = {
+    batchesToArchive.flatMap(batches => {
+      val processingBatch = new AtomicInteger(batches._2.length)
+      JobLogger.log(s"Started Processing to archive the data", Some(Map("batch_id" -> batches._1, "total_part_files_to_archive" -> processingBatch)))
+
+      // Loop through the week_num & year batch partition
+      batches._2.map((batch: BatchPartition) => {
+        val filteredDF = data.filter(
+          col("course_id") === batch.collectionId &&
+          col("batch_id") === batch.batchId &&
+          col("year") === batch.period.year &&
+          col("week_of_year") === batch.period.weekOfYear
+        ).withColumn("last_attempted_on", tsToLongUdf(col("last_attempted_on")))
+         .withColumn("updated_on", tsToLongUdf(col("updated_on")))
+         .select(columnWithOrder.head, columnWithOrder.tail: _*)
+        var archivalRequest:ArchivalRequest = getRequest(jobId, batch.collectionId, batch.batchId, List(batch.period.year, batch.period.weekOfYear))
+
+        if (archivalRequest == null) {
+          val request_data = JSONUtils.deserialize[Map[String, AnyRef]](JSONUtils.serialize(requestConfig)) ++ Map[String, Int](
+            "week" -> batch.period.weekOfYear,
+            "year"-> batch.period.year
+          )
+          archivalRequest = ArchivalRequest("", batch.batchId, batch.collectionId, Option(getReportKey), jobId, None, None, null, null, None, Option(0), JSONUtils.serialize(request_data), None)
+        }
+
+        try {
+          val urls = upload(filteredDF, JSONUtils.deserialize[Map[String, AnyRef]](JSONUtils.serialize(batch))) // Upload the archived files into blob store
+          archivalRequest.blob_url = Option(urls)
+          val metrics = ArchivalMetrics(batch, pendingWeeksOfYears = Some(processingBatch.getAndDecrement()), totalArchivedRecords = Some(filteredDF.count()), totalDeletedRecords = None)
+          JobLogger.log(s"Data is archived and Processing the remaining part files ", Some(metrics), Level.INFO)
+          markArchivalRequestAsSuccess(archivalRequest, requestConfig)
+        } catch {
+          case ex: Exception => {
+            JobLogger.log("archiveBatch: Exception with error message = " + ex.getLocalizedMessage, Some(batch), Level.ERROR)
+            markArchivalRequestAsFailed(archivalRequest, ex.getLocalizedMessage)
+          }
+        }
+      })
+    }).toList
+  }
+
+  def loadArchivedData(request: ArchivalRequest)(implicit spark: SparkSession, fc: FrameworkContext, jobConfig: JobConfig): DataFrame = {
+    ExhaustUtil.fetch(request.blob_url.get.head, "csv")
+  }
+
+  override def deleteArchivedData(requestConfig: Request, requests: Array[ArchivalRequest])(implicit spark: SparkSession, fc: FrameworkContext, config: JobConfig): List[ArchivalRequest] = {
+    requests.filter(r => r.archival_status.equals("SUCCESS") && r.deletion_status != "SUCCESS").map((request: ArchivalRequest) => {
+      deleteBatch(requestConfig, request)
+    }).toList
+  }
+
+  def deleteBatch(requestConfig: Request, request: ArchivalRequest)(implicit spark: SparkSession, fc: FrameworkContext, config: JobConfig): ArchivalRequest = {
+    try {
+      val request_data = JSONUtils.deserialize[Map[String, AnyRef]](request.request_data)
+      val batchPartition = BatchPartition(request.collection_id, request.batch_id, Period(request_data("year").asInstanceOf[Int], request_data("week").asInstanceOf[Int]))
+      val archivedData = loadArchivedData(request).select("course_id", "batch_id", "user_id", "content_id", "attempt_id")
+
+      val totalArchivedRecords: Long = archivedData.count
+      JobLogger.log(s"Deleting $totalArchivedRecords archived records only, for the year ${batchPartition.period.year} and week of year ${batchPartition.period.weekOfYear} from the DB ", None, Level.INFO)
+
+      val archivalKeyspace = requestConfig.keyspace.getOrElse(AppConf.getConfig("sunbird.courses.keyspace"))
+
+      archivedData.rdd.deleteFromCassandra(archivalKeyspace, requestConfig.archivalTable, keyColumns = SomeColumns("course_id", "batch_id", "user_id", "content_id", "attempt_id"))
+      val metrics = ArchivalMetrics(batchPartition, pendingWeeksOfYears = None, totalArchivedRecords = Some(totalArchivedRecords), totalDeletedRecords = Some(totalArchivedRecords))
+
+      JobLogger.log(s"Data is archived and Processing the remaining part files ", Some(metrics), Level.INFO)
+      markDeletionRequestAsSuccess(request, requestConfig)
+    } catch {
+      case ex: Exception => {
+        JobLogger.log("deleteBatch: Exception with error message = " + ex.getLocalizedMessage, Some(request), Level.ERROR)
+        markDeletionRequestAsFailed(request, ex.getLocalizedMessage)
+      }
+    }
+  }
+
+  def generatePeriodInData(data: DataFrame): DataFrame = {
+    data.withColumn("updated_on", to_timestamp(col(dateColumn)))
+      .withColumn("year", year(col("updated_on")))
+      .withColumn("week_of_year", weekofyear(col("updated_on")))
+      .withColumn("question", to_json(col("question")))
+  }
+}

data-products/src/main/scala/org/sunbird/analytics/archival/BaseArchivalJob.scala

@@ -0,0 +1,92 @@
+package org.sunbird.analytics.archival
+
+import com.datastax.spark.connector.cql.CassandraConnectorConf
+import org.apache.spark.SparkContext
+import org.apache.spark.sql.{DataFrame, SparkSession}
+import org.apache.spark.sql.cassandra._
+import org.apache.spark.sql.functions.udf
+import org.ekstep.analytics.framework.Level.ERROR
+import org.ekstep.analytics.framework.conf.AppConf
+import org.ekstep.analytics.framework.util.{CommonUtil, JSONUtils, JobLogger}
+import org.ekstep.analytics.framework.{FrameworkContext, IJob, JobConfig, Level}
+import org.sunbird.analytics.exhaust.BaseReportsJob
+import org.ekstep.analytics.framework.util.DatasetUtil.extensions
+import org.sunbird.analytics.archival.util.{ArchivalMetaDataStoreJob, ArchivalRequest}
+
+case class Request(archivalTable: String, keyspace: Option[String], query: Option[Map[String, AnyRef]] = Option(Map()), batchId: Option[String] = Option(""), collectionId: Option[String]=Option(""), batchFilters: Option[List[String]]=Option(List()), date: Option[String] = Option(""))
+
+trait BaseArchivalJob extends BaseReportsJob with IJob with ArchivalMetaDataStoreJob with Serializable {
+
+  val cassandraUrl = "org.apache.spark.sql.cassandra"
+  def dateColumn: String
+
+  def main(config: String)(implicit sc: Option[SparkContext] = None, fc: Option[FrameworkContext] = None): Unit = {
+    implicit val className: String = getClassName;
+    JobLogger.init(jobName)
+    JobLogger.start(s"$jobName started executing - ver3", Option(Map("config" -> config, "model" -> jobName)))
+    implicit val jobConfig: JobConfig = JSONUtils.deserialize[JobConfig](config)
+    implicit val spark: SparkSession = openSparkSession(jobConfig)
+    implicit val frameworkContext: FrameworkContext = getReportingFrameworkContext()
+
+    try {
+      val res = CommonUtil.time(execute());
+      JobLogger.end(s"$jobName completed execution", "SUCCESS", None)
+    } catch {
+      case ex: Exception => ex.printStackTrace()
+        JobLogger.log(ex.getMessage, None, ERROR);
+        JobLogger.end(jobName + " execution failed", "FAILED", Option(Map("model" -> jobName, "statusMsg" -> ex.getMessage)));
+    }
+    finally {
+      frameworkContext.closeContext();
+      spark.close()
+    }
+  }
+
+  def init()(implicit spark: SparkSession, fc: FrameworkContext, config: JobConfig): Unit = {
+    spark.setCassandraConf("LMSCluster", CassandraConnectorConf.ConnectionHostParam.option(AppConf.getConfig("sunbird.courses.cluster.host")))
+  }
+
+  def execute()(implicit spark: SparkSession, fc: FrameworkContext, config: JobConfig): Unit = {
+    val modelParams = config.modelParams.getOrElse(Map[String, Option[AnyRef]]());
+    val requestConfig = JSONUtils.deserialize[Request](JSONUtils.serialize(modelParams.getOrElse("request", Request).asInstanceOf[Map[String,AnyRef]]))
+    val mode: String = modelParams.getOrElse("mode","archive").asInstanceOf[String]
+
+    val requests = getRequests(jobId, requestConfig.batchId)
+
+    val archivalRequests = mode.toLowerCase() match {
+      case "archival" =>
+        archiveData(requestConfig, requests)
+      case "delete" =>
+        deleteArchivedData(requestConfig, requests)
+    }
+    for (archivalRequest <- archivalRequests) {
+      upsertRequest(archivalRequest)
+    }
+  }
+
+  def upload(archivedData: DataFrame, batch: Map[String,AnyRef])(implicit jobConfig: JobConfig): List[String] = {
+    val blobConfig = jobConfig.modelParams.get("blobConfig").asInstanceOf[Map[String, AnyRef]]
+    val reportPath: String = blobConfig.getOrElse("reportPath", "archived-data/").asInstanceOf[String]
+    val objectKey = AppConf.getConfig("course.metrics.cloud.objectKey")
+    val fileName = archivalFormat(batch)
+    val storageConfig = getStorageConfig(jobConfig, objectKey)
+    JobLogger.log(s"Uploading reports to blob storage", None, Level.INFO)
+    archivedData.saveToBlobStore(storageConfig, "csv", s"$reportPath$fileName-${System.currentTimeMillis()}", Option(Map("header" -> "true", "codec" -> "org.apache.hadoop.io.compress.GzipCodec")), None, fileExt=Some("csv.gz"))
+  }
+
+  // Overriding methods START:
+  def jobId: String;
+  def jobName: String;
+  def getReportPath: String;
+  def getReportKey: String;
+  def getClassName: String;
+
+  def archiveData(requestConfig: Request, requests: Array[ArchivalRequest])(implicit spark: SparkSession, fc: FrameworkContext, config: JobConfig): List[ArchivalRequest];
+  def deleteArchivedData(archivalRequest: Request, requests: Array[ArchivalRequest])(implicit spark: SparkSession, fc: FrameworkContext, config: JobConfig): List[ArchivalRequest];
+  def archivalFormat(batch: Map[String,AnyRef]): String;
+  def dataFilter(requests: Array[ArchivalRequest], dataDF: DataFrame): DataFrame;
+
+  //Overriding methods END:
+
+  val tsToLongUdf = udf[java.lang.Long, java.sql.Timestamp]((ts: java.sql.Timestamp) => if (ts != null) ts.getTime else null)
+}