-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge branch 'develop' into dateline
- Loading branch information
Showing
30 changed files
with
876 additions
and
208 deletions.
There are no files selected for viewing
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
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
38 changes: 38 additions & 0 deletions
38
geotrellis-common/src/test/scala/org/openeo/geotrelliscommon/PackageTest.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,38 @@ | ||
| package org.openeo.geotrelliscommon | ||
|
|
||
| import org.junit.jupiter.api.Assertions.{assertEquals, assertThrowsExactly, fail} | ||
| import org.junit.jupiter.api.{Test, Timeout} | ||
|
|
||
| import java.time.Duration | ||
|
|
||
| class PackageTest { | ||
| class FailedAttempt extends Exception | ||
|
|
||
| @Test | ||
| def retryForeverNumberOfAttempts(): Unit = { | ||
| var attempts = 0 | ||
|
|
||
| try { | ||
| retryForever(delay = Duration.ZERO, attempts = 3, onAttemptFailed = _ => attempts += 1) { | ||
| println("attempting...") | ||
| throw new FailedAttempt | ||
| } | ||
|
|
||
| fail("should have thrown a FailedAttempt") | ||
| } catch { | ||
| case _: FailedAttempt => | ||
| } | ||
|
|
||
| // count the number of failures to get the number of attempts | ||
| assertEquals(3, attempts) | ||
| } | ||
|
|
||
| @Test | ||
| @Timeout(5) // less than RetryForever's delay below | ||
| def retryForeverNoDelayAfterFinalFailure(): Unit = | ||
| assertThrowsExactly(classOf[FailedAttempt], () => | ||
| retryForever(delay = Duration.ofSeconds(60), attempts = 1) { | ||
| println("attempting...") | ||
| throw new FailedAttempt | ||
| }) | ||
| } |
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
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
66 changes: 66 additions & 0 deletions
66
openeo-geotrellis/src/main/java/org/openeo/sparklisteners/BatchJobProgressListener.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,66 @@ | ||
| package org.openeo.sparklisteners; | ||
|
|
||
| import org.apache.spark.executor.TaskMetrics; | ||
| import org.apache.spark.scheduler.SparkListener; | ||
| import org.apache.spark.scheduler.SparkListenerStageCompleted; | ||
| import org.apache.spark.scheduler.SparkListenerStageSubmitted; | ||
| import org.apache.spark.util.AccumulatorV2; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
| import scala.collection.Traversable; | ||
| import scala.collection.mutable.Map; | ||
|
|
||
| import java.time.Duration; | ||
|
|
||
| object BatchJobProgressListener { | ||
|
|
||
| val logger = LoggerFactory.getLogger(BatchJobProgressListener.getClass) | ||
|
|
||
| } | ||
|
|
||
| class BatchJobProgressListener extends SparkListener { | ||
|
|
||
| import BatchJobProgressListener.logger | ||
|
|
||
| override def onStageSubmitted( stageSubmitted:SparkListenerStageSubmitted):Unit = { | ||
| logger.info("Starting part of the process graph: " + stageSubmitted.stageInfo.name) | ||
| } | ||
|
|
||
| override def onStageCompleted( stageCompleted: SparkListenerStageCompleted):Unit = { | ||
| val taskMetrics = stageCompleted.stageInfo.taskMetrics | ||
|
|
||
| if(stageCompleted.stageInfo.failureReason.isDefined){ | ||
| val message = | ||
| f""" | ||
| |"A part of the process graph failed, and will be retried, the reason was: ${stageCompleted.stageInfo.failureReason.get} | ||
| |"Your job may still complete if the failure was caused by a transient error, but will take more time. A common cause of transient errors is too little executor memory (overhead). Too low executor-memory can be seen by a high 'garbage collection' time, which was: ${Duration.ofMillis(taskMetrics.jvmGCTime).toSeconds/1000.0} seconds." | ||
| |""".stripMargin | ||
| logger.warn(message); | ||
|
|
||
| }else{ | ||
| val duration = Duration.ofMillis(taskMetrics.executorRunTime) | ||
| val timeString = if(duration.toSeconds>60) { | ||
| duration.toMinutes + " minutes" | ||
| } else { | ||
| duration.toMillis.toFloat / 1000.0 + " seconds" | ||
| } | ||
| val megabytes = taskMetrics.shuffleWriteMetrics.bytesWritten.toFloat/(1024.0*1024.0) | ||
| val name = stageCompleted.stageInfo.name | ||
| logger.info(f"Finished part ${stageCompleted.stageInfo.stageId} of the process graph: ${name}.\n The total computing time was: $timeString. It produced: $megabytes%.2f MB of data."); | ||
|
|
||
| val accumulators = stageCompleted.stageInfo.accumulables; | ||
| val chunkCounts = accumulators.filter(_._2.name.get.startsWith("ChunkCount")); | ||
| if (chunkCounts.nonEmpty) { | ||
| val totalChunks = chunkCounts.head._2.value | ||
| val megapixel = totalChunks.get.asInstanceOf[Long] * 256 * 256 / (1024 * 1024) | ||
| if(taskMetrics.executorRunTime > 0) { | ||
| logger.info(f"load_collection: data was loaded with an average speed of: ${megapixel.toFloat/ duration.toSeconds.toFloat}%.3f Megapixel per second.") | ||
| }; | ||
| } | ||
| } | ||
|
|
||
|
|
||
| } | ||
|
|
||
|
|
||
| } |
Oops, something went wrong.