You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by td...@apache.org on 2017/03/28 02:01:23 UTC
spark git commit: [SPARK-19876][SS] Follow up: Refactored
BatchCommitLog to simplify logic
Repository: spark
Updated Branches:
refs/heads/master a250933c6 -> 8a6f33f04
[SPARK-19876][SS] Follow up: Refactored BatchCommitLog to simplify logic
## What changes were proposed in this pull request?
Existing logic seemingly writes null to the BatchCommitLog, even though it does additional checks to write '{}' (valid json) to the log. This PR simplifies the logic by disallowing use of `log.add(batchId, metadata)` and instead using `log.add(batchId)`. No question of specifying metadata, so no confusion related to null.
## How was this patch tested?
Existing tests pass.
Author: Tathagata Das <ta...@gmail.com>
Closes #17444 from tdas/SPARK-19876-1.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/8a6f33f0
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/8a6f33f0
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/8a6f33f0
Branch: refs/heads/master
Commit: 8a6f33f0483dcee81467e6374a796b5dbd53ea30
Parents: a250933
Author: Tathagata Das <ta...@gmail.com>
Authored: Mon Mar 27 19:04:16 2017 -0700
Committer: Tathagata Das <ta...@gmail.com>
Committed: Mon Mar 27 19:04:16 2017 -0700
----------------------------------------------------------------------
.../execution/streaming/BatchCommitLog.scala | 28 ++++++++++++--------
.../execution/streaming/HDFSMetadataLog.scala | 1 +
.../execution/streaming/StreamExecution.scala | 2 +-
3 files changed, 19 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/8a6f33f0/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/BatchCommitLog.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/BatchCommitLog.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/BatchCommitLog.scala
index fb1a4fb..a34938f 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/BatchCommitLog.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/BatchCommitLog.scala
@@ -45,33 +45,39 @@ import org.apache.spark.sql.SparkSession
class BatchCommitLog(sparkSession: SparkSession, path: String)
extends HDFSMetadataLog[String](sparkSession, path) {
+ import BatchCommitLog._
+
+ def add(batchId: Long): Unit = {
+ super.add(batchId, EMPTY_JSON)
+ }
+
+ override def add(batchId: Long, metadata: String): Boolean = {
+ throw new UnsupportedOperationException(
+ "BatchCommitLog does not take any metadata, use 'add(batchId)' instead")
+ }
+
override protected def deserialize(in: InputStream): String = {
// called inside a try-finally where the underlying stream is closed in the caller
val lines = IOSource.fromInputStream(in, UTF_8.name()).getLines()
if (!lines.hasNext) {
throw new IllegalStateException("Incomplete log file in the offset commit log")
}
- parseVersion(lines.next().trim, BatchCommitLog.VERSION)
- // read metadata
- lines.next().trim match {
- case BatchCommitLog.SERIALIZED_VOID => null
- case metadata => metadata
- }
+ parseVersion(lines.next.trim, VERSION)
+ EMPTY_JSON
}
override protected def serialize(metadata: String, out: OutputStream): Unit = {
// called inside a try-finally where the underlying stream is closed in the caller
- out.write(s"v${BatchCommitLog.VERSION}".getBytes(UTF_8))
+ out.write(s"v${VERSION}".getBytes(UTF_8))
out.write('\n')
- // write metadata or void
- out.write((if (metadata == null) BatchCommitLog.SERIALIZED_VOID else metadata)
- .getBytes(UTF_8))
+ // write metadata
+ out.write(EMPTY_JSON.getBytes(UTF_8))
}
}
object BatchCommitLog {
private val VERSION = 1
- private val SERIALIZED_VOID = "{}"
+ private val EMPTY_JSON = "{}"
}
http://git-wip-us.apache.org/repos/asf/spark/blob/8a6f33f0/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala
index 60ce642..46bfc29 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/HDFSMetadataLog.scala
@@ -106,6 +106,7 @@ class HDFSMetadataLog[T <: AnyRef : ClassTag](sparkSession: SparkSession, path:
* metadata has already been stored, this method will return `false`.
*/
override def add(batchId: Long, metadata: T): Boolean = {
+ require(metadata != null, "'null' metadata cannot written to a metadata log")
get(batchId).map(_ => false).getOrElse {
// Only write metadata when the batch has not yet been written
writeBatch(batchId, metadata)
http://git-wip-us.apache.org/repos/asf/spark/blob/8a6f33f0/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala
index 34e9262..5f54817 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/StreamExecution.scala
@@ -305,7 +305,7 @@ class StreamExecution(
if (dataAvailable) {
// Update committed offsets.
committedOffsets ++= availableOffsets
- batchCommitLog.add(currentBatchId, null)
+ batchCommitLog.add(currentBatchId)
logDebug(s"batch ${currentBatchId} committed")
// We'll increase currentBatchId after we complete processing current batch's data
currentBatchId += 1
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org