You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by va...@apache.org on 2018/12/14 00:13:03 UTC
[spark] branch master updated: [SPARK-23886][SS] Update query
status for ContinuousExecution
This is an automated email from the ASF dual-hosted git repository.
vanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 362e472 [SPARK-23886][SS] Update query status for ContinuousExecution
362e472 is described below
commit 362e472831e0609f88fdeb01d8e14badc812b0f4
Author: Gabor Somogyi <ga...@gmail.com>
AuthorDate: Thu Dec 13 16:12:55 2018 -0800
[SPARK-23886][SS] Update query status for ContinuousExecution
## What changes were proposed in this pull request?
Added query status updates to ContinuousExecution.
## How was this patch tested?
Existing unit tests + added ContinuousQueryStatusAndProgressSuite.
Closes #23095 from gaborgsomogyi/SPARK-23886.
Authored-by: Gabor Somogyi <ga...@gmail.com>
Signed-off-by: Marcelo Vanzin <va...@cloudera.com>
---
.../execution/streaming/MicroBatchExecution.scala | 6 +++
.../sql/execution/streaming/ProgressReporter.scala | 1 -
.../streaming/continuous/ContinuousExecution.scala | 6 +++
.../spark/sql/streaming/StreamingQueryStatus.scala | 6 ++-
.../ContinuousQueryStatusAndProgressSuite.scala | 55 ++++++++++++++++++++++
5 files changed, 71 insertions(+), 3 deletions(-)
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala
index 64e09ed..03beefe 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/MicroBatchExecution.scala
@@ -147,6 +147,12 @@ class MicroBatchExecution(
logInfo(s"Query $prettyIdString was stopped")
}
+ /** Begins recording statistics about query progress for a given trigger. */
+ override protected def startTrigger(): Unit = {
+ super.startTrigger()
+ currentStatus = currentStatus.copy(isTriggerActive = true)
+ }
+
/**
* Repeatedly attempts to run batches as data arrives.
*/
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ProgressReporter.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ProgressReporter.scala
index 6a22f0c..39ab702 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ProgressReporter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/ProgressReporter.scala
@@ -114,7 +114,6 @@ trait ProgressReporter extends Logging {
logDebug("Starting Trigger Calculation")
lastTriggerStartTimestamp = currentTriggerStartTimestamp
currentTriggerStartTimestamp = triggerClock.getTimeMillis()
- currentStatus = currentStatus.copy(isTriggerActive = true)
currentTriggerStartOffsets = null
currentTriggerEndOffsets = null
currentDurationsMs.clear()
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousExecution.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousExecution.scala
index 4d42428..f0859aa 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousExecution.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/continuous/ContinuousExecution.scala
@@ -118,6 +118,8 @@ class ContinuousExecution(
// For at least once, we can just ignore those reports and risk duplicates.
commitLog.getLatest() match {
case Some((latestEpochId, _)) =>
+ updateStatusMessage("Starting new streaming query " +
+ s"and getting offsets from latest epoch $latestEpochId")
val nextOffsets = offsetLog.get(latestEpochId).getOrElse {
throw new IllegalStateException(
s"Batch $latestEpochId was committed without end epoch offsets!")
@@ -129,6 +131,7 @@ class ContinuousExecution(
nextOffsets
case None =>
// We are starting this stream for the first time. Offsets are all None.
+ updateStatusMessage("Starting new streaming query")
logInfo(s"Starting new streaming query.")
currentBatchId = 0
OffsetSeq.fill(continuousSources.map(_ => null): _*)
@@ -263,6 +266,7 @@ class ContinuousExecution(
epochUpdateThread.setDaemon(true)
epochUpdateThread.start()
+ updateStatusMessage("Running")
reportTimeTaken("runContinuous") {
SQLExecution.withNewExecutionId(
sparkSessionForQuery, lastExecution) {
@@ -322,6 +326,8 @@ class ContinuousExecution(
* before this is called.
*/
def commit(epoch: Long): Unit = {
+ updateStatusMessage(s"Committing epoch $epoch")
+
assert(continuousSources.length == 1, "only one continuous source supported currently")
assert(offsetLog.get(epoch).isDefined, s"offset for epoch $epoch not reported before commit")
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryStatus.scala b/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryStatus.scala
index 9dc62b7..6ca9aac 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryStatus.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/streaming/StreamingQueryStatus.scala
@@ -28,9 +28,11 @@ import org.apache.spark.annotation.Evolving
* Reports information about the instantaneous status of a streaming query.
*
* @param message A human readable description of what the stream is currently doing.
- * @param isDataAvailable True when there is new data to be processed.
+ * @param isDataAvailable True when there is new data to be processed. Doesn't apply
+ * to ContinuousExecution where it is always false.
* @param isTriggerActive True when the trigger is actively firing, false when waiting for the
- * next trigger time.
+ * next trigger time. Doesn't apply to ContinuousExecution where it is
+ * always false.
*
* @since 2.1.0
*/
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/ContinuousQueryStatusAndProgressSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/ContinuousQueryStatusAndProgressSuite.scala
new file mode 100644
index 0000000..10bea7f
--- /dev/null
+++ b/sql/core/src/test/scala/org/apache/spark/sql/streaming/continuous/ContinuousQueryStatusAndProgressSuite.scala
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.streaming.continuous
+
+import org.apache.spark.sql.execution.streaming.StreamExecution
+import org.apache.spark.sql.execution.streaming.sources.ContinuousMemoryStream
+import org.apache.spark.sql.streaming.Trigger
+
+class ContinuousQueryStatusAndProgressSuite extends ContinuousSuiteBase {
+ test("StreamingQueryStatus - ContinuousExecution isDataAvailable and isTriggerActive " +
+ "should be false") {
+ import testImplicits._
+
+ val input = ContinuousMemoryStream[Int]
+
+ def assertStatus(stream: StreamExecution): Unit = {
+ assert(stream.status.isDataAvailable === false)
+ assert(stream.status.isTriggerActive === false)
+ }
+
+ val trigger = Trigger.Continuous(100)
+ testStream(input.toDF(), useV2Sink = true)(
+ StartStream(trigger),
+ Execute(assertStatus),
+ AddData(input, 0, 1, 2),
+ Execute(assertStatus),
+ CheckAnswer(0, 1, 2),
+ Execute(assertStatus),
+ StopStream,
+ Execute(assertStatus),
+ AddData(input, 3, 4, 5),
+ Execute(assertStatus),
+ StartStream(trigger),
+ Execute(assertStatus),
+ CheckAnswer(0, 1, 2, 3, 4, 5),
+ Execute(assertStatus),
+ StopStream,
+ Execute(assertStatus))
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org