#1371 Reassigned performance

AbsaOSS · Jul 28, 2020 · f0d53a3 · f0d53a3
1 parent cc4ef03
commit f0d53a3
Show file tree

Hide file tree

Showing 4 changed files with 7 additions and 22 deletions.
diff --git a/spark-jobs/src/main/scala/za/co/absa/enceladus/common/CommonJobExecution.scala b/spark-jobs/src/main/scala/za/co/absa/enceladus/common/CommonJobExecution.scala
@@ -85,20 +85,16 @@ trait CommonJobExecution {
     val reportVersion = getReportVersion(cmd, dataset)
     val pathCfg = getPathConfig(cmd, dataset, reportVersion)
 
-    val inputPath = getInputPath(pathCfg)
-
     validateOutputPath(fsUtils, pathCfg)
 
-    val performance = initPerformanceMeasurer(inputPath)
-
     // Enable Spline
     import za.co.absa.spline.core.SparkLineageInitializer._
     spark.enableLineageTracking()
 
     // Enable non-default persistence storage level if provided in the command line
     cmd.persistStorageLevel.foreach(Atum.setCachingStorageLevel)
 
-    PreparationResult(dataset, reportVersion, pathCfg, performance)
+    PreparationResult(dataset, reportVersion, pathCfg, new PerformanceMeasurer(spark.sparkContext.appName))
   }
 
   protected def validateOutputPath(fsUtils: FileSystemVersionUtils, pathConfig: PathConfig): Unit
@@ -111,8 +107,6 @@ trait CommonJobExecution {
     }
   }
 
-  protected def getInputPath[T](pathCfg: PathConfig): String
-
   protected def runPostProcessing[T](sourcePhase: SourcePhase, preparationResult: PreparationResult, jobCmdConfig: JobConfigParser[T])
                                     (implicit spark: SparkSession, fileSystemVersionUtils: FileSystemVersionUtils): Unit = {
     val outputPath = sourcePhase match {
@@ -247,12 +241,4 @@ trait CommonJobExecution {
         newVersion
     }
   }
-
-  private def initPerformanceMeasurer(path: String)
-                                     (implicit spark: SparkSession, fsUtils: FileSystemVersionUtils): PerformanceMeasurer = {
-    val performance = new PerformanceMeasurer(spark.sparkContext.appName)
-    val stdDirSize = fsUtils.getDirectorySize(path)
-    performance.startMeasurement(stdDirSize)
-    performance
-  }
 }
diff --git a/spark-jobs/src/main/scala/za/co/absa/enceladus/conformance/ConformanceExecution.scala b/spark-jobs/src/main/scala/za/co/absa/enceladus/conformance/ConformanceExecution.scala
@@ -52,6 +52,9 @@ trait ConformanceExecution extends CommonJobExecution {
                                       cmd: ConformanceConfigParser[T],
                                       fsUtils: FileSystemVersionUtils,
                                       spark: SparkSession): Unit = {
+    val stdDirSize = fsUtils.getDirectorySize(preparationResult.pathCfg.standardizationPath)
+    preparationResult.performance.startMeasurement(stdDirSize)
+
     log.info(s"standardization path: ${preparationResult.pathCfg.standardizationPath}")
     log.info(s"publish path: ${preparationResult.pathCfg.publishPath}")
 
@@ -91,8 +94,6 @@ trait ConformanceExecution extends CommonJobExecution {
     validateIfPathAlreadyExists(fsUtils, pathConfig.publishPath)
   }
 
-  override def getInputPath[T](pathCfg: PathConfig): String = pathCfg.standardizationPath
-
   protected def readConformanceInputData(pathCfg: PathConfig)(implicit spark: SparkSession): DataFrame = {
     spark.read.parquet(pathCfg.standardizationPath)
   }

diff --git a/...k-jobs/src/main/scala/za/co/absa/enceladus/standardization/StandardizationExecution.scala b/...k-jobs/src/main/scala/za/co/absa/enceladus/standardization/StandardizationExecution.scala
@@ -34,7 +34,7 @@ import za.co.absa.enceladus.standardization.interpreter.StandardizationInterpret
 import za.co.absa.enceladus.standardization.interpreter.stages.PlainSchemaGenerator
 import za.co.absa.enceladus.utils.fs.FileSystemVersionUtils
 import za.co.absa.enceladus.utils.modules.SourcePhase
-import za.co.absa.enceladus.utils.performance.PerformanceMetricTools
+import za.co.absa.enceladus.utils.performance.{PerformanceMeasurer, PerformanceMetricTools}
 import za.co.absa.enceladus.utils.schema.{MetadataKeys, SchemaUtils, SparkUtils}
 import za.co.absa.enceladus.utils.udf.UDFLibrary
 import za.co.absa.enceladus.utils.validation.ValidationException
@@ -52,6 +52,8 @@ trait StandardizationExecution extends CommonJobExecution {
                                           fsUtils: FileSystemVersionUtils,
                                           spark: SparkSession): StructType = {
 
+    val stdDirSize = fsUtils.getDirectorySize(preparationResult.pathCfg.rawPath)
+    preparationResult.performance.startMeasurement(stdDirSize)
     // Enable Control Framework
     import za.co.absa.atum.AtumImplicits.SparkSessionWrapper
     spark.enableControlMeasuresTracking(s"${preparationResult.pathCfg.rawPath}/_INFO")
@@ -94,8 +96,6 @@ trait StandardizationExecution extends CommonJobExecution {
     }
   }
 
-  override def getInputPath[T](pathCfg: PathConfig): String = pathCfg.rawPath
-
   override def validateOutputPath(fsUtils: FileSystemVersionUtils, pathConfig: PathConfig): Unit = {
     validateIfPathAlreadyExists(fsUtils: FileSystemVersionUtils, pathConfig.standardizationPath)
   }

diff --git a/...o/absa/enceladus/standardization_conformance/StandardizationAndConformanceExecution.scala b/...o/absa/enceladus/standardization_conformance/StandardizationAndConformanceExecution.scala
@@ -40,6 +40,4 @@ trait StandardizationAndConformanceExecution extends StandardizationExecution
     validateIfPathAlreadyExists(fsUtils, pathConfig.standardizationPath)
     validateIfPathAlreadyExists(fsUtils, pathConfig.publishPath)
   }
-
-  override def getInputPath[T](pathCfg: PathConfig): String = super[StandardizationExecution].getInputPath(pathCfg)
 }