Qualification tool - Add option to filter by minimum event log size

core/src/main/scala/com/nvidia/spark/rapids/tool/EventLogPathProcessor.scala

@@ -234,7 +234,8 @@ object EventLogPathProcessor extends Logging {
       matchlogs: Option[String],
       eventLogsPaths: List[String],
       hadoopConf: Configuration,
-      maxEventLogSize: Option[String] = None): (Seq[EventLogInfo], Seq[EventLogInfo]) = {
+      maxEventLogSize: Option[String] = None,
+      minEventLogSize: Option[String] = None): (Seq[EventLogInfo], Seq[EventLogInfo]) = {
     val logsPathNoWildCards = processWildcardsLogs(eventLogsPaths, hadoopConf)
     val logsWithTimestamp = logsPathNoWildCards.flatMap {
       case (rawPath, processedPaths) if processedPaths.isEmpty =>
@@ -252,24 +253,40 @@ object EventLogPathProcessor extends Logging {
     }.getOrElse(logsWithTimestamp)
 
     val filteredLogs = if ((filterNLogs.nonEmpty && !filterByAppCriteria(filterNLogs)) ||
-      maxEventLogSize.isDefined) {
+      maxEventLogSize.isDefined || minEventLogSize.isDefined) {
       val validMatchedLogs = matchedLogs.collect {
         case (info, Some(ts)) => info -> ts
       }
-      val filteredBySize = if (maxEventLogSize.isDefined) {
+      val filteredByMinSize = if (minEventLogSize.isDefined) {
+        val minSizeInBytes = if (StringUtils.isMemorySize(minEventLogSize.get)) {
+          // if it is memory return the bytes unit
+          StringUtils.convertMemorySizeToBytes(minEventLogSize.get)
+        } else {
+          // size is assumed to be mb
+          StringUtils.convertMemorySizeToBytes(minEventLogSize.get + "m")
+        }
+        val (matched, filtered) = validMatchedLogs.partition(info => info._2.size > minSizeInBytes)
+        logInfo(s"Filtering eventlogs by size, minimum size is ${minSizeInBytes}b. The logs " +
+          s"filtered out include: ${filtered.keys.map(_.eventLog.toString).mkString(",")}")
+        matched
+      } else {
+        validMatchedLogs
+      }
+      val filteredByMaxSize = if (maxEventLogSize.isDefined) {
         val maxSizeInBytes = if (StringUtils.isMemorySize(maxEventLogSize.get)) {
           // if it is memory return the bytes unit
           StringUtils.convertMemorySizeToBytes(maxEventLogSize.get)
         } else {
           // size is assumed to be mb
           StringUtils.convertMemorySizeToBytes(maxEventLogSize.get + "m")
         }
-        val (matched, filtered) = validMatchedLogs.partition(info => info._2.size <= maxSizeInBytes)
+        val (matched, filtered) =
+          filteredByMinSize.partition(info => info._2.size <= maxSizeInBytes)
         logInfo(s"Filtering eventlogs by size, max size is ${maxSizeInBytes}b. The logs filtered " +
           s"out include: ${filtered.keys.map(_.eventLog.toString).mkString(",")}")
         matched
       } else {
-        validMatchedLogs
+        filteredByMinSize
       }
       if (filterNLogs.nonEmpty && !filterByAppCriteria(filterNLogs)) {
         val filteredInfo = filterNLogs.get.split("-")
@@ -278,16 +295,16 @@ object EventLogPathProcessor extends Logging {
         // Before filtering based on user criteria, remove the failed event logs
         // (i.e. logs without timestamp) from the list.
         val matched = if (criteria.equals("newest")) {
-          LinkedHashMap(filteredBySize.toSeq.sortWith(_._2.timestamp > _._2.timestamp): _*)
+          LinkedHashMap(filteredByMaxSize.toSeq.sortWith(_._2.timestamp > _._2.timestamp): _*)
         } else if (criteria.equals("oldest")) {
-          LinkedHashMap(filteredBySize.toSeq.sortWith(_._2.timestamp < _._2.timestamp): _*)
+          LinkedHashMap(filteredByMaxSize.toSeq.sortWith(_._2.timestamp < _._2.timestamp): _*)
         } else {
           logError("Criteria should be either newest-filesystem or oldest-filesystem")
           Map.empty[EventLogInfo, Long]
         }
         matched.take(numberofEventLogs)
       } else {
-        filteredBySize
+        filteredByMaxSize
       }
     } else {
       matchedLogs

core/src/main/scala/com/nvidia/spark/rapids/tool/qualification/QualificationArgs.scala

@@ -88,6 +88,13 @@ Usage: java -cp rapids-4-spark-tools_2.12-<version>.jar:$SPARK_HOME/jars/*
       descr = "Filter event logs whose application start occurred within the past specified " +
         "time period. Valid time periods are min(minute),h(hours),d(days),w(weeks)," +
         "m(months). If a period is not specified it defaults to days.")
+  val minEventLogSize: ScallopOption[String] =
+    opt[String](required = false,
+      descr = "Process only application event logs whose size is greater than the size " +
+        "specified. Valid units of size are b(bytes),k(kilobytes),m(megabytes),g(gigabytes). " +
+        "If no units are specified, the size is assumed to be m. Note, this does not support " +
+        "event log rolling which puts multiple event logs for the same application into a " +
+        "single directory.")
   val maxEventLogSize: ScallopOption[String] =
     opt[String](required = false,
       descr = "Process only application event logs whose size is less than or equal to the size " +

core/src/main/scala/com/nvidia/spark/rapids/tool/qualification/QualificationMain.scala

@@ -50,6 +50,7 @@ object QualificationMain extends Logging {
 
     val eventlogPaths = appArgs.eventlog()
     val filterN = appArgs.filterCriteria
+    val minEventLogSize = appArgs.minEventLogSize.toOption
     val maxEventLogSize = appArgs.maxEventLogSize.toOption
     val matchEventLogs = appArgs.matchEventLogs
     val outputDirectory = appArgs.outputDirectory().stripSuffix("/")
@@ -87,7 +88,8 @@ object QualificationMain extends Logging {
     }
 
     val (eventLogFsFiltered, allEventLogs) = EventLogPathProcessor.processAllPaths(
-      filterN.toOption, matchEventLogs.toOption, eventlogPaths, hadoopConf, maxEventLogSize)
+      filterN.toOption, matchEventLogs.toOption, eventlogPaths, hadoopConf,
+      maxEventLogSize, minEventLogSize)
 
     val filteredLogs = if (argsContainsAppFilters(appArgs)) {
       val appFilter = new AppFilterImpl(numOutputRows, hadoopConf, timeout, nThreads)
@@ -122,9 +124,10 @@ object QualificationMain extends Logging {
   def argsContainsFSFilters(appArgs: QualificationArgs): Boolean = {
     val filterCriteria = appArgs.filterCriteria.toOption
     val maxEventLogSize = appArgs.maxEventLogSize.toOption
+    val minEventLogSize = appArgs.minEventLogSize.toOption
     appArgs.matchEventLogs.isSupplied ||
       (filterCriteria.isDefined && filterCriteria.get.endsWith("-filesystem")) ||
-      maxEventLogSize.isDefined
+      maxEventLogSize.isDefined || minEventLogSize.isDefined
   }
 
   def argsContainsAppFilters(appArgs: QualificationArgs): Boolean = {

user_tools/src/spark_rapids_pytools/resources/qualification-conf.yaml

@@ -85,6 +85,7 @@ sparkRapids:
       - no-html-report
       - m
       - match-event-logs
+      - min-event-log-size
       - max-event-log-size
       - max-sql-desc-length
       - ml-functions