mediative / amadou Goto Github PK

Motivation

One context to rule them all.

Input

• amadou/core/src/main/scala/com.mediative.amadou/monitoring/MessagingSystem.scala

  Lines 23 to 29 in 1b1eca3

  	case class ProcessContext(
  	jobId: String,
  	processingDate: DateInterval,
  	processId: String = java.util.UUID.randomUUID().toString,
  	startTime: Long = System.currentTimeMillis) {
  	def duration: Long = System.currentTimeMillis - startTime
  	}

• amadou/core/src/main/scala/com.mediative.amadou/SparkRunner.scala

  Lines 101 to 133 in 1b1eca3

  	class Context[+I](job: Job, processContext: ProcessContext, spark: SparkSession, date: DateInterval, value: I)
  	extends Stage.Context(spark, date, value) {
  	override def withValue[U](value: U) = new Context(job, processContext, spark, date, value)
  	override def run[T](stage: Stage[I, T], result: => T): Stage.Result[T] = {
  	def runStage(callCount: Int): Stage.Result[T] = {
  	logger.info(s"[$processContext] Running stage ${stage.name} try #$callCount")
  	counters.foreach(_.clear())
  	messaging.publishStageStarting(processContext, stage.name)
  	Try(result) match {
  	case v @ Success(_) =>
  	messaging.publishStageComplete(processContext, stage.name)
  	messaging.publishMetrics(processContext, stage.name, collectMetrics())
  	v
  	case Failure(failure) =>
  	if (callCount >= job.maxRetries) {
  	logger.error(s"[$processContext] Giving up after ${job.maxRetries} retries", failure)
  	messaging.publishStageFailed(processContext, stage.name, failure)
  	messaging.publishProcessFailed(processContext, failure)
  	throw failure
  	} else {
  	logger.error(s"[$processContext] Will retry stage $stage in ${job.delayBetweenRetries}", failure)
  	messaging.publishStageRetrying(processContext, stage.name)
  	Thread.sleep(job.delayBetweenRetries.toMillis)
  	runStage(callCount + 1)
  	}
  	}
  	}
  	runStage(1)
  	}
  	}

Output

• A single component which maintains the current context of a job/stage.

Motivation

The naming convention from Apache Beam is very good and amadou should use it. This would also allow to make certain parts easier to test by removing use of hard coded processing date (e.g. Day.today).

Input

• Current context class

Output

• Context class updated with date renamed to eventDate and a new processingDate member.

#18

Motivation

Each stage automatically emits the recordsProcessed metric which may lead to zero values for stages that do not do any Spark reads. Avoid the zero values to not skew the dashboard which does a mean on the values returned for a given day.

Input

The sparkRecordsRead counter.

Output

No recordsProcessed custom metric is emitted for stages that does no reads.

Motivation: make <job>/run exit back to sbt immediately.

Failure to resolve a Kafka broker will crash a job

Caused by: org.apache.kafka.common.config.ConfigException: DNS resolution failed for url in bootstrap.servers: broker-1.kafka.mesos:6000
	at org.apache.kafka.common.utils.ClientUtils.parseAndValidateAddresses(ClientUtils.java:38)
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:189)
	at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:129)
	at com.mediative.amadou.monitoring.KafkaMessagingSystem.<init>(KafkaMessagingSystem.scala:26)
	at com.mediative.amadou.monitoring.MessagingSystem$.create(MessagingSystem.scala:33)
	at com.mediative.amadou.SparkRunner.<init>(SparkRunner.scala:140)
	at com.mediative.mpn.brain.dataflow.jobs.national.National$.<init>(National.scala:14)
	at com.mediative.mpn.brain.dataflow.jobs.national.National$.<clinit>(National.scala)
	... 1 more

Motivation

Ensure that all data in a dataset is written to BQ.

Input

• job.getStatistics.getLoad.getOutputRows

Output

• BigQueryClient.load fails if output rows != dataset rows.

Test

• Test with mock BQ client that triggers the failure

From emitted log:

17/04/26 14:15:55 ERROR AAA$: [ProcessContext(aaa,2017-04-26,98fc8de3-259c-4c3c-865c-8e9dc1003372,1493215211215)] Will retry stage com.mediative.amadou.Stage$$anon$3@2cfbeac4 in 20 minutes