forked from pflooky/data-caterer
-
-
Notifications
You must be signed in to change notification settings - Fork 5
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #86 from data-catering/plan-results
Add in plan results, clean up imports
- Loading branch information
Showing
43 changed files
with
455 additions
and
308 deletions.
There are no files selected for viewing
4 changes: 2 additions & 2 deletions
4
...c/main/scala/io/github/datacatering/datacaterer/api/DataCatererConfigurationBuilder.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
2 changes: 1 addition & 1 deletion
2
api/src/main/scala/io/github/datacatering/datacaterer/api/MetadataSourceBuilder.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
2 changes: 1 addition & 1 deletion
2
api/src/main/scala/io/github/datacatering/datacaterer/api/model/MetadataSourceModels.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
162 changes: 162 additions & 0 deletions
162
api/src/main/scala/io/github/datacatering/datacaterer/api/model/ResultModels.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,162 @@ | ||
| package io.github.datacatering.datacaterer.api.model | ||
|
|
||
| import io.github.datacatering.datacaterer.api.model.Constants.DEFAULT_DATA_SOURCE_NAME | ||
| import io.github.datacatering.datacaterer.api.util.ConfigUtil.cleanseOptions | ||
| import io.github.datacatering.datacaterer.api.util.ResultWriterUtil.getSuccessSymbol | ||
|
|
||
| import java.time.{Duration, LocalDateTime} | ||
| import scala.math.BigDecimal.RoundingMode | ||
|
|
||
|
|
||
| case class DataSourceResultSummary( | ||
| name: String, | ||
| numRecords: Long, | ||
| isSuccess: Boolean, | ||
| dataSourceResults: List[DataSourceResult] | ||
| ) | ||
|
|
||
| case class DataSourceResult( | ||
| name: String = DEFAULT_DATA_SOURCE_NAME, | ||
| task: Task = Task(), | ||
| step: Step = Step(), | ||
| sinkResult: SinkResult = SinkResult(), | ||
| batchNum: Int = 0 | ||
| ) { | ||
|
|
||
| def summarise: List[String] = { | ||
| val format = sinkResult.format | ||
| val isSuccess = getSuccessSymbol(sinkResult.isSuccess) | ||
| val numRecords = sinkResult.count.toString | ||
| List(name, format, isSuccess, numRecords) | ||
| } | ||
|
|
||
| def jsonSummary: Map[String, Any] = { | ||
| Map( | ||
| "name" -> name, | ||
| "options" -> step.options, | ||
| "isSuccess" -> sinkResult.isSuccess, | ||
| "numRecords" -> sinkResult.count | ||
| ) | ||
| } | ||
| } | ||
|
|
||
| case class TaskResultSummary( | ||
| task: Task, | ||
| numRecords: Long, | ||
| isSuccess: Boolean, | ||
| stepResults: List[StepResultSummary] | ||
| ) | ||
|
|
||
| case class StepResultSummary( | ||
| step: Step, | ||
| numRecords: Long, | ||
| isSuccess: Boolean, | ||
| dataSourceResults: List[DataSourceResult] | ||
| ) | ||
|
|
||
| case class SinkResult( | ||
| name: String = DEFAULT_DATA_SOURCE_NAME, | ||
| format: String = "json", | ||
| saveMode: String = "append", | ||
| options: Map[String, String] = Map(), | ||
| count: Long = -1, | ||
| isSuccess: Boolean = true, | ||
| sample: Array[String] = Array(), | ||
| startTime: LocalDateTime = LocalDateTime.now(), | ||
| endTime: LocalDateTime = LocalDateTime.now(), | ||
| generatedMetadata: Array[Field] = Array(), | ||
| exception: Option[Throwable] = None | ||
| ) { | ||
|
|
||
| def durationInSeconds: Long = Duration.between(startTime, endTime).toSeconds | ||
| } | ||
|
|
||
|
|
||
| case class ValidationConfigResult( | ||
| name: String = "default_validation_result", | ||
| description: String = "Validation result for data sources", | ||
| dataSourceValidationResults: List[DataSourceValidationResult] = List(), | ||
| startTime: LocalDateTime = LocalDateTime.now(), | ||
| endTime: LocalDateTime = LocalDateTime.now() | ||
| ) { | ||
| def durationInSeconds: Long = Duration.between(startTime, endTime).toSeconds | ||
|
|
||
| def summarise: List[String] = { | ||
| val validationRes = dataSourceValidationResults.flatMap(_.validationResults) | ||
| if (validationRes.nonEmpty) { | ||
| val (numSuccess, successRate, isSuccess) = baseSummary(validationRes) | ||
| val successRateVisual = s"$numSuccess/${validationRes.size} ($successRate%)" | ||
| List(name, description, getSuccessSymbol(isSuccess), successRateVisual) | ||
| } else List() | ||
| } | ||
|
|
||
| def jsonSummary: Map[String, Any] = { | ||
| val validationRes = dataSourceValidationResults.flatMap(dsv => | ||
| dsv.validationResults.map(v => (dsv.dataSourceName, dsv.options, v)) | ||
| ) | ||
| if (validationRes.nonEmpty) { | ||
| val (numSuccess, successRate, isSuccess) = baseSummary(validationRes.map(_._3)) | ||
| val errorMap = validationRes.filter(vr => !vr._3.isSuccess).map(res => { | ||
| val validationDetails = res._3.validation.toOptions.map(v => (v.head, v.last)).toMap | ||
| Map( | ||
| "dataSourceName" -> res._1, | ||
| "options" -> cleanseOptions(res._2), | ||
| "validation" -> validationDetails, | ||
| "numErrors" -> res._3.numErrors, | ||
| "sampleErrorValues" -> res._3.sampleErrorValues.getOrElse(Array()) | ||
| ) | ||
| }) | ||
| val baseValidationMap = Map( | ||
| "name" -> name, | ||
| "description" -> description, | ||
| "isSuccess" -> isSuccess, | ||
| "numSuccess" -> numSuccess, | ||
| "numValidations" -> validationRes.size, | ||
| "successRate" -> successRate | ||
| ) | ||
| if (errorMap.nonEmpty) { | ||
| baseValidationMap ++ Map("errorValidations" -> errorMap) | ||
| } else baseValidationMap | ||
| } else Map() | ||
| } | ||
|
|
||
| private def baseSummary(validationRes: List[ValidationResult]): (Int, BigDecimal, Boolean) = { | ||
| val validationSuccess = validationRes.map(_.isSuccess) | ||
| val numSuccess = validationSuccess.count(x => x) | ||
| val successRate = BigDecimal(numSuccess.toDouble / validationRes.size * 100).setScale(2, RoundingMode.HALF_UP) | ||
| val isSuccess = validationSuccess.forall(x => x) | ||
| (numSuccess, successRate, isSuccess) | ||
| } | ||
| } | ||
|
|
||
| case class DataSourceValidationResult( | ||
| dataSourceName: String = "default_data_source", | ||
| options: Map[String, String] = Map(), | ||
| validationResults: List[ValidationResult] = List() | ||
| ) | ||
|
|
||
| case class ValidationResult( | ||
| validation: Validation = ExpressionValidation(), | ||
| isSuccess: Boolean = true, | ||
| numErrors: Long = 0, | ||
| total: Long = 0, | ||
| sampleErrorValues: Option[Array[Map[String, Any]]] = None | ||
| ) | ||
|
|
||
| object ValidationResult { | ||
| def fromValidationWithBaseResult(validation: Validation, validationResult: ValidationResult): ValidationResult = { | ||
| ValidationResult(validation, validationResult.isSuccess, validationResult.numErrors, validationResult.total, validationResult.sampleErrorValues) | ||
| } | ||
| } | ||
|
|
||
| case class PlanResults( | ||
| plan: Plan, | ||
| generationResult: List[DataSourceResult], | ||
| validationResults: List[ValidationConfigResult] | ||
| ) | ||
|
|
||
| case class PlanRunSummary( | ||
| plan: Plan, | ||
| tasks: List[Task], | ||
| validations: List[ValidationConfiguration] | ||
| ) |
2 changes: 1 addition & 1 deletion
2
...ng/datacaterer/core/util/ConfigUtil.scala → ...ing/datacaterer/api/util/ConfigUtil.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
2 changes: 1 addition & 1 deletion
2
...acaterer/core/util/ResultWriterUtil.scala → ...tacaterer/api/util/ResultWriterUtil.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
46 changes: 46 additions & 0 deletions
46
...ain/scala/io/github/datacatering/datacaterer/core/activity/PlanRunPostPlanProcessor.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,46 @@ | ||
| package io.github.datacatering.datacaterer.core.activity | ||
|
|
||
| import io.github.datacatering.datacaterer.api.model.{DataCatererConfiguration, DataSourceResult, Plan, PlanResults, ValidationConfigResult} | ||
| import io.github.datacatering.datacaterer.core.listener.SparkRecordListener | ||
| import io.github.datacatering.datacaterer.core.plan.PostPlanProcessor | ||
| import io.github.datacatering.datacaterer.core.util.ManagementUtil.getDataCatererManagementUrl | ||
| import io.github.datacatering.datacaterer.core.util.ObjectMapperUtil | ||
| import org.apache.log4j.Logger | ||
| import org.asynchttpclient.AsyncHttpClient | ||
| import org.asynchttpclient.Dsl.asyncHttpClient | ||
|
|
||
| import scala.compat.java8.FutureConverters.CompletionStageOps | ||
| import scala.concurrent.ExecutionContext.Implicits.global | ||
| import scala.util.{Failure, Success} | ||
|
|
||
| class PlanRunPostPlanProcessor(val dataCatererConfiguration: DataCatererConfiguration) extends PostPlanProcessor { | ||
|
|
||
| override val enabled: Boolean = dataCatererConfiguration.flagsConfig.enableTrackActivity | ||
|
|
||
| private val LOGGER = Logger.getLogger(getClass.getName) | ||
| private val dataCatererManagementUrl = getDataCatererManagementUrl | ||
| private val http: AsyncHttpClient = asyncHttpClient | ||
|
|
||
| override def apply( | ||
| plan: Plan, | ||
| sparkRecordListener: SparkRecordListener, | ||
| generationResult: List[DataSourceResult], | ||
| validationResults: List[ValidationConfigResult] | ||
| ): Unit = { | ||
| val planResults = PlanResults(plan, generationResult, validationResults) | ||
| val jsonBody = ObjectMapperUtil.jsonObjectMapper.writeValueAsString(planResults) | ||
| val url = s"$dataCatererManagementUrl/plan/finish" | ||
| val prepareRequest = http.prepare("POST", url) | ||
| .setBody(jsonBody) | ||
|
|
||
| val futureResp = prepareRequest.execute().toCompletableFuture.toScala | ||
| futureResp.onComplete { | ||
| case Success(_) => | ||
| LOGGER.debug(s"Successfully posted run results, url=$url") | ||
| http.close() | ||
| case Failure(exception) => | ||
| LOGGER.debug(s"Failed to post run results, url=$url", exception) | ||
| http.close() | ||
| } | ||
| } | ||
| } |
44 changes: 44 additions & 0 deletions
44
...main/scala/io/github/datacatering/datacaterer/core/activity/PlanRunPrePlanProcessor.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| package io.github.datacatering.datacaterer.core.activity | ||
|
|
||
| import io.github.datacatering.datacaterer.api.model.{DataCatererConfiguration, Plan, PlanRunSummary, Task, ValidationConfiguration} | ||
| import io.github.datacatering.datacaterer.core.plan.PrePlanProcessor | ||
| import io.github.datacatering.datacaterer.core.util.ManagementUtil.getDataCatererManagementUrl | ||
| import io.github.datacatering.datacaterer.core.util.ObjectMapperUtil | ||
| import org.apache.log4j.Logger | ||
| import org.asynchttpclient.AsyncHttpClient | ||
| import org.asynchttpclient.Dsl.asyncHttpClient | ||
|
|
||
| import scala.compat.java8.FutureConverters.CompletionStageOps | ||
| import scala.concurrent.ExecutionContext.Implicits.global | ||
| import scala.util.{Failure, Success} | ||
|
|
||
| class PlanRunPrePlanProcessor(val dataCatererConfiguration: DataCatererConfiguration) extends PrePlanProcessor { | ||
|
|
||
| override val enabled: Boolean = dataCatererConfiguration.flagsConfig.enableTrackActivity | ||
|
|
||
| private val LOGGER = Logger.getLogger(getClass.getName) | ||
| private val dataCatererManagementUrl = getDataCatererManagementUrl | ||
| private val http: AsyncHttpClient = asyncHttpClient | ||
|
|
||
| override def apply( | ||
| plan: Plan, | ||
| tasks: List[Task], | ||
| validations: List[ValidationConfiguration] | ||
| ): Unit = { | ||
| val planRunSummary = PlanRunSummary(plan, tasks, validations) | ||
| val jsonBody = ObjectMapperUtil.jsonObjectMapper.writeValueAsString(planRunSummary) | ||
| val url = s"$dataCatererManagementUrl/plan/start" | ||
| val prepareRequest = http.prepare("POST", url) | ||
| .setBody(jsonBody) | ||
|
|
||
| val futureResp = prepareRequest.execute().toCompletableFuture.toScala | ||
| futureResp.onComplete { | ||
| case Success(_) => | ||
| LOGGER.debug(s"Successfully posted run start, url=$url") | ||
| http.close() | ||
| case Failure(exception) => | ||
| LOGGER.debug(s"Failed to post run start, url=$url", exception) | ||
| http.close() | ||
| } | ||
| } | ||
| } |
3 changes: 1 addition & 2 deletions
3
app/src/main/scala/io/github/datacatering/datacaterer/core/alert/AlertProcessor.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.