You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by jerryshao <gi...@git.apache.org> on 2016/06/05 03:51:08 UTC
[GitHub] spark pull request #13513: [SPARK-15698][SQL][Streaming] Add the ability to ...
GitHub user jerryshao opened a pull request:
https://github.com/apache/spark/pull/13513
[SPARK-15698][SQL][Streaming] Add the ability to remove the old MetadataLog in FileStreamSource
## What changes were proposed in this pull request?
Current `metadataLog` in `FileStreamSource` will add a checkpoint file in each batch but do not have the ability to remove/compact, which will lead to large number of small files when running for a long time. So here propose to compact the old logs into one file. This method is quite similar to `FileStreamSinkLog` but simpler.
## How was this patch tested?
Unit test added.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/jerryshao/apache-spark SPARK-15698
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/13513.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #13513
----
commit 2ed1115966fdf8b6a8fba990599b230a04e00649
Author: jerryshao <ss...@hortonworks.com>
Date: 2016-06-02T03:14:48Z
Add the ability to remove the old MetadataLog in FileStreamSource
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on the issue:
https://github.com/apache/spark/pull/13513
@zsxwing @frreiss thanks a lot for your comments.
I think the semantics of `FileStreamSource.getBatch(start: Option[Offset], end: Offset)` still keeps the same, since I overrided the `get` method in `FileStreamSourceLog` and filter out some compacted data.
Yes it could be slow to get a batch where it happens to be a compact batch. I think we could have 2 solutions:
1. doing compact on the next of latest metadata file (as what I did before), then this will help most of the scenarios in `FileStreamSource`.
2. We could put the data in this patch at beginning when doing compaction, so we don't need to scan the whole file to get this batch's metadata.
Both two solutions need extra works, what do you think?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/65245/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #13513: [SPARK-15698][SQL][Streaming] Add the ability to ...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on a diff in the pull request:
https://github.com/apache/spark/pull/13513#discussion_r65945436
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala ---
@@ -129,3 +131,86 @@ class FileStreamSource(
override def toString: String = s"FileStreamSource[$qualifiedBasePath]"
}
+
+class FileStreamSourceLog(sparkSession: SparkSession, path: String)
+ extends HDFSMetadataLog[Seq[String]](sparkSession, path) {
+
+ // Configurations about metadata compaction
+ private val compactInterval = sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL)
+ require(compactInterval > 0,
+ s"Please set ${SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL.key} (was $compactInterval) to a " +
+ s"positive value.")
+
+ private val fileCleanupDelayMs = sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_CLEANUP_DELAY)
+
+ private val isDeletingExpiredLog = sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_DELETION)
+
+ private var compactBatchId: Long = -1L
+
+ private def isCompactionBatch(batchId: Long, compactInterval: Long): Boolean = {
+ batchId % compactInterval == 0
+ }
+
+ override def add(batchId: Long, metadata: Seq[String]): Boolean = {
+ if (isCompactionBatch(batchId, compactInterval)) {
+ compactMetadataLog(batchId - 1)
+ }
+
+ super.add(batchId, metadata)
+ }
+
+ private def compactMetadataLog(batchId: Long): Unit = {
+ // read out compact metadata and merge with new metadata.
+ val batches = super.get(Some(compactBatchId), Some(batchId))
+ val totalMetadata = batches.flatMap(_._2)
+ if (totalMetadata.isEmpty) {
+ return
+ }
+
+ // Remove old compact metadata file and rewrite.
+ val renamedPath = new Path(path, s".${batchId.toString}-${UUID.randomUUID.toString}.tmp")
+ fileManager.rename(batchIdToPath(batchId), renamedPath)
+
+ var isSuccess = false
+ try {
+ isSuccess = super.add(batchId, totalMetadata)
+ } catch {
+ case NonFatal(e) => isSuccess = false
--- End diff --
Yeah, it is not necessary, I will remove it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on the issue:
https://github.com/apache/spark/pull/13513
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #65245 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65245/consoleFull)** for PR 13513 at commit [`f179349`](https://github.com/apache/spark/commit/f1793498a9625dc8d31039cd8e9a684611dddf23).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `abstract class CompactibleFileStreamLog[T: ClassTag](`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on the issue:
https://github.com/apache/spark/pull/13513
@jerryshao here is a test case to show the issue about `getBatch`:
```Scala
test("getBatch") {
withTempDirs { case (src, tmp) =>
withSQLConf(
SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL.key -> "2",
// Force deleting the old logs
SQLConf.FILE_SOURCE_LOG_CLEANUP_DELAY.key -> "1"
) {
val fileStream = createFileStream("text", src.getCanonicalPath)
val filtered = fileStream.filter($"value" contains "keep")
testStream(filtered)(
AddTextFileData("keep1", src, tmp),
CheckAnswer("keep1"),
AddTextFileData("keep2", src, tmp),
CheckAnswer("keep1", "keep2"),
AddTextFileData("keep3", src, tmp),
CheckAnswer("keep1", "keep2", "keep3"),
AssertOnQuery("check getBatch") { execution: StreamExecution =>
val _sources = PrivateMethod[Seq[Source]]('sources)
val fileSource =
(execution invokePrivate _sources()).head.asInstanceOf[FileStreamSource]
assert(fileSource.getBatch(None, LongOffset(2)).as[String].collect() ===
List("keep1", "keep2", "keep3"))
assert(fileSource.getBatch(Some(LongOffset(0)), LongOffset(2)).as[String].collect() ===
List("keep2", "keep3"))
assert(fileSource.getBatch(Some(LongOffset(1)), LongOffset(2)).as[String].collect() ===
List("keep3"))
}
)
}
}
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/65547/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on the issue:
https://github.com/apache/spark/pull/13513
Jenkins, retest this please.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on the issue:
https://github.com/apache/spark/pull/13513
@zsxwing , thanks a lot for your comments, I did several refactorings:
1. Abstract and consolidate `FileStreamSinkLog` and `FileStreamSourceLog`, now they share same code path to do compaction.
2. Change `FileStreamSourceLog` to use json format instead of binary coding, to add the compatibility and flexibility for future extension.
3. Improve the logics to fetch all metadata logs, now if compact log is existed, only scan compact log.
Please help to review again, thanks a lot.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on the issue:
https://github.com/apache/spark/pull/13513
Thanks a lot @zsxwing and @frreiss for your comments.
For the slow scan problem of compact batch. Originally I planned to to not merge the latest batch as I did before, also suggested above. but with several different tries it is hard to implement with small changes. So for now I still choose the same implementation with a simple cache layer to overcome this problem, the basic compaction algorithm is still the same as `FileStreamSinkLog`. I think it is easier to maintain.
For the problem of semantics broken. I realized that it is really a problem, but current code didn't touch it. So I changed to scan the compacted batch files to retrieve missing batches. It is a little time-consuming, and the current logic of `FileStreamSource` will not touch this part.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/60060/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on the issue:
https://github.com/apache/spark/pull/13513
Sure, I will change the code.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #13513: [SPARK-15698][SQL][Streaming] Add the ability to ...
Posted by jaceklaskowski <gi...@git.apache.org>.
Github user jaceklaskowski commented on a diff in the pull request:
https://github.com/apache/spark/pull/13513#discussion_r65825474
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala ---
@@ -129,3 +131,86 @@ class FileStreamSource(
override def toString: String = s"FileStreamSource[$qualifiedBasePath]"
}
+
+class FileStreamSourceLog(sparkSession: SparkSession, path: String)
+ extends HDFSMetadataLog[Seq[String]](sparkSession, path) {
+
+ // Configurations about metadata compaction
+ private val compactInterval = sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL)
+ require(compactInterval > 0,
+ s"Please set ${SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL.key} (was $compactInterval) to a " +
+ s"positive value.")
+
+ private val fileCleanupDelayMs = sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_CLEANUP_DELAY)
+
+ private val isDeletingExpiredLog = sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_DELETION)
+
+ private var compactBatchId: Long = -1L
+
+ private def isCompactionBatch(batchId: Long, compactInterval: Long): Boolean = {
+ batchId % compactInterval == 0
+ }
+
+ override def add(batchId: Long, metadata: Seq[String]): Boolean = {
+ if (isCompactionBatch(batchId, compactInterval)) {
+ compactMetadataLog(batchId - 1)
+ }
+
+ super.add(batchId, metadata)
+ }
+
+ private def compactMetadataLog(batchId: Long): Unit = {
+ // read out compact metadata and merge with new metadata.
+ val batches = super.get(Some(compactBatchId), Some(batchId))
+ val totalMetadata = batches.flatMap(_._2)
+ if (totalMetadata.isEmpty) {
+ return
+ }
+
+ // Remove old compact metadata file and rewrite.
+ val renamedPath = new Path(path, s".${batchId.toString}-${UUID.randomUUID.toString}.tmp")
+ fileManager.rename(batchIdToPath(batchId), renamedPath)
+
+ var isSuccess = false
+ try {
+ isSuccess = super.add(batchId, totalMetadata)
+ } catch {
+ case NonFatal(e) => isSuccess = false
--- End diff --
Why are you setting `isSuccess` to `false` since it's `false` already?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/65244/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #60071 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/60071/consoleFull)** for PR 13513 at commit [`798c450`](https://github.com/apache/spark/commit/798c4502bb657e986b1dcce09983bd44c7a2f342).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #13513: [SPARK-15698][SQL][Streaming] Add the ability to ...
Posted by jaceklaskowski <gi...@git.apache.org>.
Github user jaceklaskowski commented on a diff in the pull request:
https://github.com/apache/spark/pull/13513#discussion_r65825524
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala ---
@@ -529,7 +529,28 @@ object SQLConf {
.internal()
.doc("How long in milliseconds a file is guaranteed to be visible for all readers.")
.timeConf(TimeUnit.MILLISECONDS)
- .createWithDefault(60 * 1000L) // 10 minutes
+ .createWithDefault(60 * 10 * 1000L) // 10 minutes
+
+ val FILE_SOURCE_LOG_DELETION = SQLConfigBuilder("spark.sql.streaming.fileSource.log.deletion")
+ .internal()
+ .doc("Whether to delete the expired log files in file stream source.")
+ .booleanConf
+ .createWithDefault(true)
+
+ val FILE_SOURCE_LOG_COMPACT_INTERVAL =
+ SQLConfigBuilder("spark.sql.streaming.fileSource.log.compactInterval")
+ .internal()
+ .doc("Number of log files after which all the previous files " +
+ "are compacted into the next log file.")
+ .intConf
+ .createWithDefault(10)
+
+ val FILE_SOURCE_LOG_CLEANUP_DELAY =
+ SQLConfigBuilder("spark.sql.streaming.fileSource.log.cleanupDelay")
+ .internal()
+ .doc("How long in milliseconds a file is guaranteed to be visible for all readers.")
+ .timeConf(TimeUnit.MILLISECONDS)
+ .createWithDefault(60 * 10 * 1000L) // 10 minutes
--- End diff --
A nitpick but think it'd be easier to "decode" - `10 * 60 * 1000L`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/65628/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #13513: [SPARK-15698][SQL][Streaming] Add the ability to ...
Posted by jaceklaskowski <gi...@git.apache.org>.
Github user jaceklaskowski commented on a diff in the pull request:
https://github.com/apache/spark/pull/13513#discussion_r65825480
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala ---
@@ -129,3 +131,86 @@ class FileStreamSource(
override def toString: String = s"FileStreamSource[$qualifiedBasePath]"
}
+
+class FileStreamSourceLog(sparkSession: SparkSession, path: String)
+ extends HDFSMetadataLog[Seq[String]](sparkSession, path) {
+
+ // Configurations about metadata compaction
+ private val compactInterval = sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL)
+ require(compactInterval > 0,
+ s"Please set ${SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL.key} (was $compactInterval) to a " +
+ s"positive value.")
+
+ private val fileCleanupDelayMs = sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_CLEANUP_DELAY)
+
+ private val isDeletingExpiredLog = sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_DELETION)
+
+ private var compactBatchId: Long = -1L
+
+ private def isCompactionBatch(batchId: Long, compactInterval: Long): Boolean = {
+ batchId % compactInterval == 0
+ }
+
+ override def add(batchId: Long, metadata: Seq[String]): Boolean = {
+ if (isCompactionBatch(batchId, compactInterval)) {
+ compactMetadataLog(batchId - 1)
+ }
+
+ super.add(batchId, metadata)
+ }
+
+ private def compactMetadataLog(batchId: Long): Unit = {
+ // read out compact metadata and merge with new metadata.
+ val batches = super.get(Some(compactBatchId), Some(batchId))
+ val totalMetadata = batches.flatMap(_._2)
+ if (totalMetadata.isEmpty) {
+ return
+ }
+
+ // Remove old compact metadata file and rewrite.
+ val renamedPath = new Path(path, s".${batchId.toString}-${UUID.randomUUID.toString}.tmp")
+ fileManager.rename(batchIdToPath(batchId), renamedPath)
+
+ var isSuccess = false
+ try {
+ isSuccess = super.add(batchId, totalMetadata)
+ } catch {
+ case NonFatal(e) => isSuccess = false
+ } finally {
+ if (!isSuccess) {
+ // Rollback to the previous status if compaction is failed.
--- End diff --
s/status/state ?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #13513: [SPARK-15698][SQL][Streaming] Add the ability to ...
Posted by jaceklaskowski <gi...@git.apache.org>.
Github user jaceklaskowski commented on a diff in the pull request:
https://github.com/apache/spark/pull/13513#discussion_r65825440
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala ---
@@ -129,3 +131,86 @@ class FileStreamSource(
override def toString: String = s"FileStreamSource[$qualifiedBasePath]"
}
+
+class FileStreamSourceLog(sparkSession: SparkSession, path: String)
+ extends HDFSMetadataLog[Seq[String]](sparkSession, path) {
+
+ // Configurations about metadata compaction
+ private val compactInterval = sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL)
+ require(compactInterval > 0,
+ s"Please set ${SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL.key} (was $compactInterval) to a " +
--- End diff --
I'd move `(was $compactInterval)` at the end of the message.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #65631 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65631/consoleFull)** for PR 13513 at commit [`84d3d27`](https://github.com/apache/spark/commit/84d3d27490556dc1de4e4bce3b6b19a75691f52e).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #65244 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65244/consoleFull)** for PR 13513 at commit [`c2aad87`](https://github.com/apache/spark/commit/c2aad87ba012c41a0f4ef6290401e6789f2c9ed6).
* This patch **fails to build**.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* ` case class FileEntry(path: String, timestamp: Timestamp, action: String = ADD_ACTION)`
* ` class FileStreamSourceLog(sparkSession: SparkSession, path: String)`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #65547 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65547/consoleFull)** for PR 13513 at commit [`be1abfa`](https://github.com/apache/spark/commit/be1abfa0e902fca3ed945bfbb6e0573909d55e2b).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/65368/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #13513: [SPARK-15698][SQL][Streaming] Add the ability to ...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/13513#discussion_r78630727
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/CompactibleFileStreamLog.scala ---
@@ -0,0 +1,249 @@
+/*
+ * 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.execution.streaming
+
+import java.io.IOException
+import java.nio.charset.StandardCharsets.UTF_8
+import java.util.concurrent.TimeUnit
+
+import scala.reflect.ClassTag
+
+import org.apache.hadoop.fs.{Path, PathFilter}
+
+import org.apache.spark.sql.SparkSession
+
+/**
+ * An abstract class for compactible metadata logs. It will write one log file for each batch.
+ * The first line of the log file is the version number, and there are multiple JSON lines
+ * following.
+ *
+ * As reading from many small files is usually pretty slow, also too many
+ * small files in one folder will mess the FS, [[CompactibleFileStreamLog]] will
+ * compact log files every 10 batches by default into a big file. When
+ * doing a compaction, it will read all old log files and merge them with the new batch.
+ */
+abstract class CompactibleFileStreamLog[T: ClassTag](
+ sparkSession: SparkSession,
+ path: String)
+ extends HDFSMetadataLog[Array[T]](sparkSession, path) {
+
+ import CompactibleFileStreamLog._
+
+ /**
+ * If we delete the old files after compaction at once, there is a race condition in S3: other
+ * processes may see the old files are deleted but still cannot see the compaction file using
+ * "list". The `allFiles` handles this by looking for the next compaction file directly, however,
+ * a live lock may happen if the compaction happens too frequently: one processing keeps deleting
+ * old files while another one keeps retrying. Setting a reasonable cleanup delay could avoid it.
+ */
+ protected val fileCleanupDelayMs = TimeUnit.MINUTES.toMillis(10)
+
+ protected val isDeletingExpiredLog = true
+
+ protected val compactInterval = 10
+
+ /**
+ * Serialize the data into encoded string.
+ */
+ protected def serializeData(t: T): String
+
+ /**
+ * Deserialize the string into data object.
+ */
+ protected def deserializeData(encodedString: String): T
+
+ /**
+ * Filter out the unwanted logs, by default it filters out nothing, inherited class could
+ * override this method to do filtering.
+ */
+ protected def compactLogs(oldLogs: Seq[T], newLogs: Seq[T]): Seq[T] = {
+ oldLogs ++ newLogs
+ }
+
+ override def batchIdToPath(batchId: Long): Path = {
+ if (isCompactionBatch(batchId, compactInterval)) {
+ new Path(metadataPath, s"$batchId$COMPACT_FILE_SUFFIX")
+ } else {
+ new Path(metadataPath, batchId.toString)
+ }
+ }
+
+ override def pathToBatchId(path: Path): Long = {
+ getBatchIdFromFileName(path.getName)
+ }
+
+ override def isBatchFile(path: Path): Boolean = {
+ try {
+ getBatchIdFromFileName(path.getName)
+ true
+ } catch {
+ case _: NumberFormatException => false
+ }
+ }
+
+ override def serialize(logData: Array[T]): Array[Byte] = {
+ (VERSION +: logData.map(serializeData)).mkString("\n").getBytes(UTF_8)
--- End diff --
Could you make VERSION be a constructor parameter in order to support to change source or sink format separately?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/65631/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #13513: [SPARK-15698][SQL][Streaming] Add the ability to ...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on a diff in the pull request:
https://github.com/apache/spark/pull/13513#discussion_r79093102
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSourceLog.scala ---
@@ -0,0 +1,133 @@
+/*
+ * 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.execution.streaming
+
+import scala.collection.mutable
+
+import org.json4s.NoTypeHints
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.execution.streaming.FileStreamSource.FileEntry
+import org.apache.spark.sql.internal.SQLConf
+
+class FileStreamSourceLog(
+ metadataLogVersion: String,
+ sparkSession: SparkSession,
+ path: String)
+ extends CompactibleFileStreamLog[FileEntry](metadataLogVersion, sparkSession, path) {
+
+ import CompactibleFileStreamLog._
+
+ // Configurations about metadata compaction
+ protected override val compactInterval =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL)
+ require(compactInterval > 0,
+ s"Please set ${SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL.key} (was $compactInterval) to a " +
+ s"positive value.")
+
+ protected override val fileCleanupDelayMs =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_CLEANUP_DELAY)
+
+ protected override val isDeletingExpiredLog =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_DELETION)
+
+ private implicit val formats = Serialization.formats(NoTypeHints)
+
+ // A fixed size log cache to cache the file entries belong to the compaction batch. It is used
+ // to avoid scanning the compacted log file to retrieve it's own batch data.
+ private val cacheSize = compactInterval
+ private val fileEntryCache = new mutable.LinkedHashMap[Long, Array[FileEntry]]
+
+ private def updateCache(batchId: Long, logs: Array[FileEntry]): Unit = {
+ if (fileEntryCache.size >= cacheSize) {
+ fileEntryCache.drop(1)
--- End diff --
I see, sorry for this issue.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #65246 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65246/consoleFull)** for PR 13513 at commit [`31340b5`](https://github.com/apache/spark/commit/31340b58ffa7c46c2d9666569d5694bb23cc6144).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #65628 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65628/consoleFull)** for PR 13513 at commit [`bddbc7f`](https://github.com/apache/spark/commit/bddbc7f8e1563000ea4a9dcad07c92e34c24199f).
* This patch **fails to build**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #13513: [SPARK-15698][SQL][Streaming] Add the ability to ...
Posted by frreiss <gi...@git.apache.org>.
Github user frreiss commented on a diff in the pull request:
https://github.com/apache/spark/pull/13513#discussion_r76499068
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala ---
@@ -129,3 +131,86 @@ class FileStreamSource(
override def toString: String = s"FileStreamSource[$qualifiedBasePath]"
}
+
+class FileStreamSourceLog(sparkSession: SparkSession, path: String)
+ extends HDFSMetadataLog[Seq[String]](sparkSession, path) {
+
+ // Configurations about metadata compaction
+ private val compactInterval = sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL)
+ require(compactInterval > 0,
+ s"Please set ${SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL.key} (was $compactInterval) to a " +
+ s"positive value.")
+
+ private val fileCleanupDelayMs = sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_CLEANUP_DELAY)
+
+ private val isDeletingExpiredLog = sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_DELETION)
+
+ private var compactBatchId: Long = -1L
+
+ private def isCompactionBatch(batchId: Long, compactInterval: Long): Boolean = {
+ batchId % compactInterval == 0
+ }
+
+ override def add(batchId: Long, metadata: Seq[String]): Boolean = {
+ if (isCompactionBatch(batchId, compactInterval)) {
+ compactMetadataLog(batchId - 1)
+ }
+
+ super.add(batchId, metadata)
+ }
+
+ private def compactMetadataLog(batchId: Long): Unit = {
+ // read out compact metadata and merge with new metadata.
+ val batches = super.get(Some(compactBatchId), Some(batchId))
+ val totalMetadata = batches.flatMap(_._2)
+ if (totalMetadata.isEmpty) {
+ return
+ }
+
+ // Remove old compact metadata file and rewrite.
+ val renamedPath = new Path(path, s".${batchId.toString}-${UUID.randomUUID.toString}.tmp")
+ fileManager.rename(batchIdToPath(batchId), renamedPath)
+
+ var isSuccess = false
+ try {
+ isSuccess = super.add(batchId, totalMetadata)
+ } catch {
+ case NonFatal(e) => isSuccess = false
+ } finally {
+ if (!isSuccess) {
+ // Rollback to the previous status if compaction is failed.
--- End diff --
This rollback code will not execute if the process exits during a compaction operation. You will need cleanup code in the class constructor to handle that case.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on the issue:
https://github.com/apache/spark/pull/13513
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/60000/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #60060 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/60060/consoleFull)** for PR 13513 at commit [`2ed1115`](https://github.com/apache/spark/commit/2ed1115966fdf8b6a8fba990599b230a04e00649).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/60071/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by frreiss <gi...@git.apache.org>.
Github user frreiss commented on the issue:
https://github.com/apache/spark/pull/13513
Ah, now I fully understand @zsxwing's earlier comment about the semantics of the semantics of `Source.getBatch()`. Those semantics have a design flaw; see the email thread I started at http://apache-spark-developers-list.1001551.n3.nabble.com/Source-API-requires-unbounded-distributed-storage-tt18551.html. Basically, it's impossible to implement a Source to the written API spec without keeping unbounded state. I have an open PR to fix this problem at https://github.com/apache/spark/pull/14553.
In the short run, I think that @jerryshao's changes here are ok with respect to `Source.getBatch`. The approach in this PR will work as long as the internal structure of the `StreamExecution` class doesn't change and as long as Spark does not have to recover from an outage longer than the compaction interval. The recent changes to `FileInputStream` under SPARK-17165 (https://github.com/apache/spark/pull/14728) have the same problem, and those changes are already committed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #13513: [SPARK-15698][SQL][Streaming] Add the ability to ...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/13513
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #65368 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65368/consoleFull)** for PR 13513 at commit [`cb4194e`](https://github.com/apache/spark/commit/cb4194e58e73bfcbe27cfce84440308d18e70d84).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #65365 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65365/consoleFull)** for PR 13513 at commit [`f9a4bcb`](https://github.com/apache/spark/commit/f9a4bcb75cac9d1cd11ee519eeffa7c1354fb233).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #60000 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/60000/consoleFull)** for PR 13513 at commit [`2ed1115`](https://github.com/apache/spark/commit/2ed1115966fdf8b6a8fba990599b230a04e00649).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `class FileStreamSourceLog(sparkSession: SparkSession, path: String)`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #65628 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65628/consoleFull)** for PR 13513 at commit [`bddbc7f`](https://github.com/apache/spark/commit/bddbc7f8e1563000ea4a9dcad07c92e34c24199f).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #13513: [SPARK-15698][SQL][Streaming] Add the ability to ...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/13513#discussion_r79518445
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSourceLog.scala ---
@@ -0,0 +1,132 @@
+/*
+ * 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.execution.streaming
+
+import java.util.{LinkedHashMap => JLinkedHashMap}
+import java.util.Map.Entry
+
+import scala.collection.mutable
+
+import org.json4s.NoTypeHints
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.execution.streaming.FileStreamSource.FileEntry
+import org.apache.spark.sql.internal.SQLConf
+
+class FileStreamSourceLog(
+ metadataLogVersion: String,
+ sparkSession: SparkSession,
+ path: String)
+ extends CompactibleFileStreamLog[FileEntry](metadataLogVersion, sparkSession, path) {
+
+ import CompactibleFileStreamLog._
+
+ // Configurations about metadata compaction
+ protected override val compactInterval =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL)
+ require(compactInterval > 0,
+ s"Please set ${SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL.key} (was $compactInterval) to a " +
+ s"positive value.")
+
+ protected override val fileCleanupDelayMs =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_CLEANUP_DELAY)
+
+ protected override val isDeletingExpiredLog =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_DELETION)
+
+ private implicit val formats = Serialization.formats(NoTypeHints)
+
+ // A fixed size log entry cache to cache the file entries belong to the compaction batch. It is
+ // used to avoid scanning the compacted log file to retrieve it's own batch data.
+ private val cacheSize = compactInterval
+ private val fileEntryCache = new JLinkedHashMap[Long, Array[FileEntry]] {
+ override def removeEldestEntry(eldest: Entry[Long, Array[FileEntry]]): Boolean = {
+ size() > cacheSize
+ }
+ }
+
+ protected override def serializeData(data: FileEntry): String = {
+ Serialization.write(data)
+ }
+
+ protected override def deserializeData(encodedString: String): FileEntry = {
+ Serialization.read[FileEntry](encodedString)
+ }
+
+ def compactLogs(logs: Seq[FileEntry]): Seq[FileEntry] = {
+ logs
+ }
+
+ override def add(batchId: Long, logs: Array[FileEntry]): Boolean = {
+ if (super.add(batchId, logs) && isCompactionBatch(batchId, compactInterval)) {
--- End diff --
This is wrong. If `super.add(batchId, logs)` is false, then we should always return `false`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #65245 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65245/consoleFull)** for PR 13513 at commit [`f179349`](https://github.com/apache/spark/commit/f1793498a9625dc8d31039cd8e9a684611dddf23).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on the issue:
https://github.com/apache/spark/pull/13513
Just noticed that `FileStreamSource.getBatch(start: Option[Offset], end: Offset)` is broken in this PR. `start` could be an arbitrary offset.
I think we need to store `batchId` with its file paths together in the metadata log. `FileStreamSource.getBatch(start: Option[Offset], end: Offset)` could be very slow when all batches are in the same file because we need to parse the whole file to get the mapping from `batchId` to `files`. However, in most cases, `FileStreamSource.getBatch` only queries the latest batch, so if we don't compact the latest metadata file, we can make it pretty fast by reading one small file for most of cases. When recovering from failure, the performance of `FileStreamSource.getBatch` doesn't really matter.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/65246/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on the issue:
https://github.com/apache/spark/pull/13513
LGTM. Thanks! Merging to master and 2.0.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #65368 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65368/consoleFull)** for PR 13513 at commit [`cb4194e`](https://github.com/apache/spark/commit/cb4194e58e73bfcbe27cfce84440308d18e70d84).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #65365 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65365/consoleFull)** for PR 13513 at commit [`f9a4bcb`](https://github.com/apache/spark/commit/f9a4bcb75cac9d1cd11ee519eeffa7c1354fb233).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `class FileStreamSinkLog(`
* ` case class FileEntry(path: String, timestamp: Timestamp, batchId: Long = NOT_SET)`
* `class FileStreamSourceLog(`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/65365/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #65246 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65246/consoleFull)** for PR 13513 at commit [`31340b5`](https://github.com/apache/spark/commit/31340b58ffa7c46c2d9666569d5694bb23cc6144).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #65631 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65631/consoleFull)** for PR 13513 at commit [`84d3d27`](https://github.com/apache/spark/commit/84d3d27490556dc1de4e4bce3b6b19a75691f52e).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on the issue:
https://github.com/apache/spark/pull/13513
@jerryshao the approach seems good to me. Could you refactor the codes to avoid copying codes from FileStreamSinkLog? It's hard to maintain duplicated codes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on the issue:
https://github.com/apache/spark/pull/13513
@tdas @zsxwing , what is your comment about this PR? Thanks a lot.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #13513: [SPARK-15698][SQL][Streaming] Add the ability to ...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on a diff in the pull request:
https://github.com/apache/spark/pull/13513#discussion_r79530152
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSourceLog.scala ---
@@ -0,0 +1,132 @@
+/*
+ * 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.execution.streaming
+
+import java.util.{LinkedHashMap => JLinkedHashMap}
+import java.util.Map.Entry
+
+import scala.collection.mutable
+
+import org.json4s.NoTypeHints
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.execution.streaming.FileStreamSource.FileEntry
+import org.apache.spark.sql.internal.SQLConf
+
+class FileStreamSourceLog(
+ metadataLogVersion: String,
+ sparkSession: SparkSession,
+ path: String)
+ extends CompactibleFileStreamLog[FileEntry](metadataLogVersion, sparkSession, path) {
+
+ import CompactibleFileStreamLog._
+
+ // Configurations about metadata compaction
+ protected override val compactInterval =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL)
+ require(compactInterval > 0,
+ s"Please set ${SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL.key} (was $compactInterval) to a " +
+ s"positive value.")
+
+ protected override val fileCleanupDelayMs =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_CLEANUP_DELAY)
+
+ protected override val isDeletingExpiredLog =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_DELETION)
+
+ private implicit val formats = Serialization.formats(NoTypeHints)
+
+ // A fixed size log entry cache to cache the file entries belong to the compaction batch. It is
+ // used to avoid scanning the compacted log file to retrieve it's own batch data.
+ private val cacheSize = compactInterval
+ private val fileEntryCache = new JLinkedHashMap[Long, Array[FileEntry]] {
+ override def removeEldestEntry(eldest: Entry[Long, Array[FileEntry]]): Boolean = {
+ size() > cacheSize
+ }
+ }
+
+ protected override def serializeData(data: FileEntry): String = {
+ Serialization.write(data)
+ }
+
+ protected override def deserializeData(encodedString: String): FileEntry = {
+ Serialization.read[FileEntry](encodedString)
+ }
+
+ def compactLogs(logs: Seq[FileEntry]): Seq[FileEntry] = {
+ logs
+ }
+
+ override def add(batchId: Long, logs: Array[FileEntry]): Boolean = {
+ if (super.add(batchId, logs) && isCompactionBatch(batchId, compactInterval)) {
--- End diff --
yes, you're right, I will fix it.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/13513
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #60071 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/60071/consoleFull)** for PR 13513 at commit [`798c450`](https://github.com/apache/spark/commit/798c4502bb657e986b1dcce09983bd44c7a2f342).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #60000 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/60000/consoleFull)** for PR 13513 at commit [`2ed1115`](https://github.com/apache/spark/commit/2ed1115966fdf8b6a8fba990599b230a04e00649).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #65547 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65547/consoleFull)** for PR 13513 at commit [`be1abfa`](https://github.com/apache/spark/commit/be1abfa0e902fca3ed945bfbb6e0573909d55e2b).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #13513: [SPARK-15698][SQL][Streaming] Add the ability to ...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/13513#discussion_r79668559
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSinkLog.scala ---
@@ -79,213 +76,46 @@ object SinkFileStatus {
* When the reader uses `allFiles` to list all files, this method only returns the visible files
* (drops the deleted files).
*/
-class FileStreamSinkLog(sparkSession: SparkSession, path: String)
- extends HDFSMetadataLog[Array[SinkFileStatus]](sparkSession, path) {
-
- import FileStreamSinkLog._
+class FileStreamSinkLog(
+ metadataLogVersion: String,
+ sparkSession: SparkSession,
+ path: String)
+ extends CompactibleFileStreamLog[SinkFileStatus](metadataLogVersion, sparkSession, path) {
private implicit val formats = Serialization.formats(NoTypeHints)
- /**
- * If we delete the old files after compaction at once, there is a race condition in S3: other
- * processes may see the old files are deleted but still cannot see the compaction file using
- * "list". The `allFiles` handles this by looking for the next compaction file directly, however,
- * a live lock may happen if the compaction happens too frequently: one processing keeps deleting
- * old files while another one keeps retrying. Setting a reasonable cleanup delay could avoid it.
- */
- private val fileCleanupDelayMs = sparkSession.sessionState.conf.fileSinkLogCleanupDelay
+ protected override val fileCleanupDelayMs =
--- End diff --
I just noticed some conflicts here. Could you submit a follow up PR to use the previous `sparkSession.sessionState.conf.fileSinkLogCleanupDelay`? Same as the other confs. This only exists in master branch, so we don't need to fix branch 2.0.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on the issue:
https://github.com/apache/spark/pull/13513
@frreiss SPARK-17165 (#14728) uses `SeenFilesMap.lastPurgeTimestamp` to ignore files. when recovering from failure, `SeenFilesMap.lastPurgeTimestamp` will be set via the files in the metadata log. File paths not stored in the memory but older than `SeenFilesMap.lastPurgeTimestamp` won't be processed. Therefore, it doesn't need to store unbounded state.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by frreiss <gi...@git.apache.org>.
Github user frreiss commented on the issue:
https://github.com/apache/spark/pull/13513
You could just move the metadata deletion logic from FileStreamSinkLog into CompactibleFileStreamLog. Then FileStreamSource could issue DELETE log records for files that are older than `FileStreamSource.lastPurgeTimestamp`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #60060 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/60060/consoleFull)** for PR 13513 at commit [`2ed1115`](https://github.com/apache/spark/commit/2ed1115966fdf8b6a8fba990599b230a04e00649).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `class FileStreamSourceLog(sparkSession: SparkSession, path: String)`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on the issue:
https://github.com/apache/spark/pull/13513
Still saw the following errors in the unit-test log:
```Scala
16/09/13 07:41:18.817 shuffle-server-466-7 WARN TransportChannelHandler: Exception in connection from /127.0.0.1:36871
io.netty.util.internal.OutOfDirectMemoryError: failed to allocate 16777216 byte(s) of direct memory (used: 3120562176, max: 3132096512)
at io.netty.util.internal.PlatformDependent.incrementMemoryCounter(PlatformDependent.java:528)
at io.netty.util.internal.PlatformDependent.allocateDirectNoCleaner(PlatformDependent.java:482)
at io.netty.buffer.PoolArena$DirectArena.allocateDirect(PoolArena.java:688)
at io.netty.buffer.PoolArena$DirectArena.newChunk(PoolArena.java:677)
at io.netty.buffer.PoolArena.allocateNormal(PoolArena.java:237)
at io.netty.buffer.PoolArena.allocate(PoolArena.java:213)
at io.netty.buffer.PoolArena.allocate(PoolArena.java:141)
at io.netty.buffer.PooledByteBufAllocator.newDirectBuffer(PooledByteBufAllocator.java:271)
at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:177)
at io.netty.buffer.AbstractByteBufAllocator.directBuffer(AbstractByteBufAllocator.java:168)
at io.netty.buffer.AbstractByteBufAllocator.ioBuffer(AbstractByteBufAllocator.java:129)
at io.netty.channel.AdaptiveRecvByteBufAllocator$HandleImpl.allocate(AdaptiveRecvByteBufAllocator.java:104)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:117)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:611)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:552)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:466)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:438)
at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:140)
at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144)
at java.lang.Thread.run(Thread.java:745)
```
Seems the error cannot be reported correctly in Jenkins.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/13513
**[Test build #65244 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65244/consoleFull)** for PR 13513 at commit [`c2aad87`](https://github.com/apache/spark/commit/c2aad87ba012c41a0f4ef6290401e6789f2c9ed6).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #13513: [SPARK-15698][SQL][Streaming] Add the ability to ...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on a diff in the pull request:
https://github.com/apache/spark/pull/13513#discussion_r79099764
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSourceLog.scala ---
@@ -0,0 +1,133 @@
+/*
+ * 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.execution.streaming
+
+import scala.collection.mutable
+
+import org.json4s.NoTypeHints
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.execution.streaming.FileStreamSource.FileEntry
+import org.apache.spark.sql.internal.SQLConf
+
+class FileStreamSourceLog(
+ metadataLogVersion: String,
+ sparkSession: SparkSession,
+ path: String)
+ extends CompactibleFileStreamLog[FileEntry](metadataLogVersion, sparkSession, path) {
+
+ import CompactibleFileStreamLog._
+
+ // Configurations about metadata compaction
+ protected override val compactInterval =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL)
+ require(compactInterval > 0,
+ s"Please set ${SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL.key} (was $compactInterval) to a " +
+ s"positive value.")
+
+ protected override val fileCleanupDelayMs =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_CLEANUP_DELAY)
+
+ protected override val isDeletingExpiredLog =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_DELETION)
+
+ private implicit val formats = Serialization.formats(NoTypeHints)
+
+ // A fixed size log cache to cache the file entries belong to the compaction batch. It is used
+ // to avoid scanning the compacted log file to retrieve it's own batch data.
+ private val cacheSize = compactInterval
+ private val fileEntryCache = new mutable.LinkedHashMap[Long, Array[FileEntry]]
+
+ private def updateCache(batchId: Long, logs: Array[FileEntry]): Unit = {
+ if (fileEntryCache.size >= cacheSize) {
+ fileEntryCache.drop(1)
+ }
+
+ fileEntryCache.put(batchId, logs)
+ }
+
+ protected override def serializeData(data: FileEntry): String = {
+ Serialization.write(data)
+ }
+
+ protected override def deserializeData(encodedString: String): FileEntry = {
+ Serialization.read[FileEntry](encodedString)
+ }
+
+ def compactLogs(logs: Seq[FileEntry]): Seq[FileEntry] = {
+ logs
+ }
+
+ override def add(batchId: Long, logs: Array[FileEntry]): Boolean = {
+ if (super.add(batchId, logs) && isCompactionBatch(batchId, compactInterval)) {
+ updateCache(batchId, logs)
+ true
+ } else if (!isCompactionBatch(batchId, compactInterval)) {
+ true
+ } else {
+ false
+ }
+ }
+
+ override def get(startId: Option[Long], endId: Option[Long]): Array[(Long, Array[FileEntry])] = {
+ val startBatchId = startId.getOrElse(0L)
+ val endBatchId = getLatest().map(_._1).getOrElse(0L)
+
+ val (existedBatches, removedBatches) = (startBatchId to endBatchId).map { id =>
+ if (isCompactionBatch(id, compactInterval) && fileEntryCache.contains(id)) {
+ (id, Some(fileEntryCache(id)))
+ } else {
+ val logs = super.get(id).map(_.filter(_.batchId == id))
+ (id, logs)
+ }
+ }.partition(_._2.isDefined)
+
+ // The below code may only be happened when original metadata log file has been removed, so we
+ // have to get the batch from latest compacted log file. This is quite time-consuming and may
+ // not be happened in the current FileStreamSource code path, since we only fetch the
+ // latest metadata log file.
+ val searchKeys = removedBatches.map(_._1)
+ val retrievedBatches = if (searchKeys.nonEmpty) {
+ logWarning(s"Get batches from removed files, this is unexpected in the current code path!!!")
+ val latestBatchId = getLatest().map(_._1).getOrElse(-1L)
+ if (latestBatchId < 0) {
+ Map.empty[Long, Option[Array[FileEntry]]]
+ } else {
+ val latestCompactedBatchId = getAllValidBatches(latestBatchId, compactInterval)(0)
+ val allLogs = new mutable.HashMap[Long, mutable.ArrayBuffer[FileEntry]]
+
+ super.get(latestCompactedBatchId).foreach { entries =>
--- End diff --
I think here in the parent class `CompactibleFileStreamLog` we assume that metadata type should be `Array[T]`, which is suitable for both file source and sink log currently. If we change to use `FileSourceLogEntry`, the base class should be `T` rather then `Array[T]`, which will make the two inherited class divergent.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #13513: [SPARK-15698][SQL][Streaming] Add the ability to ...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/13513#discussion_r79518675
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSourceLog.scala ---
@@ -0,0 +1,133 @@
+/*
+ * 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.execution.streaming
+
+import scala.collection.mutable
+
+import org.json4s.NoTypeHints
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.execution.streaming.FileStreamSource.FileEntry
+import org.apache.spark.sql.internal.SQLConf
+
+class FileStreamSourceLog(
+ metadataLogVersion: String,
+ sparkSession: SparkSession,
+ path: String)
+ extends CompactibleFileStreamLog[FileEntry](metadataLogVersion, sparkSession, path) {
+
+ import CompactibleFileStreamLog._
+
+ // Configurations about metadata compaction
+ protected override val compactInterval =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL)
+ require(compactInterval > 0,
+ s"Please set ${SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL.key} (was $compactInterval) to a " +
+ s"positive value.")
+
+ protected override val fileCleanupDelayMs =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_CLEANUP_DELAY)
+
+ protected override val isDeletingExpiredLog =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_DELETION)
+
+ private implicit val formats = Serialization.formats(NoTypeHints)
+
+ // A fixed size log cache to cache the file entries belong to the compaction batch. It is used
+ // to avoid scanning the compacted log file to retrieve it's own batch data.
+ private val cacheSize = compactInterval
+ private val fileEntryCache = new mutable.LinkedHashMap[Long, Array[FileEntry]]
+
+ private def updateCache(batchId: Long, logs: Array[FileEntry]): Unit = {
+ if (fileEntryCache.size >= cacheSize) {
+ fileEntryCache.drop(1)
+ }
+
+ fileEntryCache.put(batchId, logs)
+ }
+
+ protected override def serializeData(data: FileEntry): String = {
+ Serialization.write(data)
+ }
+
+ protected override def deserializeData(encodedString: String): FileEntry = {
+ Serialization.read[FileEntry](encodedString)
+ }
+
+ def compactLogs(logs: Seq[FileEntry]): Seq[FileEntry] = {
+ logs
+ }
+
+ override def add(batchId: Long, logs: Array[FileEntry]): Boolean = {
+ if (super.add(batchId, logs) && isCompactionBatch(batchId, compactInterval)) {
+ updateCache(batchId, logs)
+ true
+ } else if (!isCompactionBatch(batchId, compactInterval)) {
+ true
+ } else {
+ false
+ }
+ }
+
+ override def get(startId: Option[Long], endId: Option[Long]): Array[(Long, Array[FileEntry])] = {
+ val startBatchId = startId.getOrElse(0L)
+ val endBatchId = getLatest().map(_._1).getOrElse(0L)
+
+ val (existedBatches, removedBatches) = (startBatchId to endBatchId).map { id =>
+ if (isCompactionBatch(id, compactInterval) && fileEntryCache.contains(id)) {
+ (id, Some(fileEntryCache(id)))
+ } else {
+ val logs = super.get(id).map(_.filter(_.batchId == id))
+ (id, logs)
+ }
+ }.partition(_._2.isDefined)
+
+ // The below code may only be happened when original metadata log file has been removed, so we
+ // have to get the batch from latest compacted log file. This is quite time-consuming and may
+ // not be happened in the current FileStreamSource code path, since we only fetch the
+ // latest metadata log file.
+ val searchKeys = removedBatches.map(_._1)
+ val retrievedBatches = if (searchKeys.nonEmpty) {
+ logWarning(s"Get batches from removed files, this is unexpected in the current code path!!!")
+ val latestBatchId = getLatest().map(_._1).getOrElse(-1L)
+ if (latestBatchId < 0) {
+ Map.empty[Long, Option[Array[FileEntry]]]
+ } else {
+ val latestCompactedBatchId = getAllValidBatches(latestBatchId, compactInterval)(0)
+ val allLogs = new mutable.HashMap[Long, mutable.ArrayBuffer[FileEntry]]
+
+ super.get(latestCompactedBatchId).foreach { entries =>
--- End diff --
Thanks for your clarifying.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #13513: [SPARK-15698][SQL][Streaming] Add the ability to ...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/13513#discussion_r79027018
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSourceLog.scala ---
@@ -0,0 +1,133 @@
+/*
+ * 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.execution.streaming
+
+import scala.collection.mutable
+
+import org.json4s.NoTypeHints
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.execution.streaming.FileStreamSource.FileEntry
+import org.apache.spark.sql.internal.SQLConf
+
+class FileStreamSourceLog(
+ metadataLogVersion: String,
+ sparkSession: SparkSession,
+ path: String)
+ extends CompactibleFileStreamLog[FileEntry](metadataLogVersion, sparkSession, path) {
+
+ import CompactibleFileStreamLog._
+
+ // Configurations about metadata compaction
+ protected override val compactInterval =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL)
+ require(compactInterval > 0,
+ s"Please set ${SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL.key} (was $compactInterval) to a " +
+ s"positive value.")
+
+ protected override val fileCleanupDelayMs =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_CLEANUP_DELAY)
+
+ protected override val isDeletingExpiredLog =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_DELETION)
+
+ private implicit val formats = Serialization.formats(NoTypeHints)
+
+ // A fixed size log cache to cache the file entries belong to the compaction batch. It is used
+ // to avoid scanning the compacted log file to retrieve it's own batch data.
+ private val cacheSize = compactInterval
+ private val fileEntryCache = new mutable.LinkedHashMap[Long, Array[FileEntry]]
+
+ private def updateCache(batchId: Long, logs: Array[FileEntry]): Unit = {
+ if (fileEntryCache.size >= cacheSize) {
+ fileEntryCache.drop(1)
+ }
+
+ fileEntryCache.put(batchId, logs)
+ }
+
+ protected override def serializeData(data: FileEntry): String = {
+ Serialization.write(data)
+ }
+
+ protected override def deserializeData(encodedString: String): FileEntry = {
+ Serialization.read[FileEntry](encodedString)
+ }
+
+ def compactLogs(logs: Seq[FileEntry]): Seq[FileEntry] = {
+ logs
+ }
+
+ override def add(batchId: Long, logs: Array[FileEntry]): Boolean = {
+ if (super.add(batchId, logs) && isCompactionBatch(batchId, compactInterval)) {
+ updateCache(batchId, logs)
+ true
+ } else if (!isCompactionBatch(batchId, compactInterval)) {
+ true
+ } else {
+ false
+ }
+ }
+
+ override def get(startId: Option[Long], endId: Option[Long]): Array[(Long, Array[FileEntry])] = {
+ val startBatchId = startId.getOrElse(0L)
+ val endBatchId = getLatest().map(_._1).getOrElse(0L)
+
+ val (existedBatches, removedBatches) = (startBatchId to endBatchId).map { id =>
+ if (isCompactionBatch(id, compactInterval) && fileEntryCache.contains(id)) {
+ (id, Some(fileEntryCache(id)))
+ } else {
+ val logs = super.get(id).map(_.filter(_.batchId == id))
+ (id, logs)
+ }
+ }.partition(_._2.isDefined)
+
+ // The below code may only be happened when original metadata log file has been removed, so we
+ // have to get the batch from latest compacted log file. This is quite time-consuming and may
+ // not be happened in the current FileStreamSource code path, since we only fetch the
+ // latest metadata log file.
+ val searchKeys = removedBatches.map(_._1)
+ val retrievedBatches = if (searchKeys.nonEmpty) {
+ logWarning(s"Get batches from removed files, this is unexpected in the current code path!!!")
+ val latestBatchId = getLatest().map(_._1).getOrElse(-1L)
+ if (latestBatchId < 0) {
+ Map.empty[Long, Option[Array[FileEntry]]]
+ } else {
+ val latestCompactedBatchId = getAllValidBatches(latestBatchId, compactInterval)(0)
+ val allLogs = new mutable.HashMap[Long, mutable.ArrayBuffer[FileEntry]]
+
+ super.get(latestCompactedBatchId).foreach { entries =>
--- End diff --
How about we use the following class for FileStreamSource?
```
case class FileSourceLogEntry(batchId: Long, Seq[FileEntry])
```
I think this will make the codes here simpler.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #13513: [SPARK-15698][SQL][Streaming] Add the ability to ...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on a diff in the pull request:
https://github.com/apache/spark/pull/13513#discussion_r79747197
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSinkLog.scala ---
@@ -79,213 +76,46 @@ object SinkFileStatus {
* When the reader uses `allFiles` to list all files, this method only returns the visible files
* (drops the deleted files).
*/
-class FileStreamSinkLog(sparkSession: SparkSession, path: String)
- extends HDFSMetadataLog[Array[SinkFileStatus]](sparkSession, path) {
-
- import FileStreamSinkLog._
+class FileStreamSinkLog(
+ metadataLogVersion: String,
+ sparkSession: SparkSession,
+ path: String)
+ extends CompactibleFileStreamLog[SinkFileStatus](metadataLogVersion, sparkSession, path) {
private implicit val formats = Serialization.formats(NoTypeHints)
- /**
- * If we delete the old files after compaction at once, there is a race condition in S3: other
- * processes may see the old files are deleted but still cannot see the compaction file using
- * "list". The `allFiles` handles this by looking for the next compaction file directly, however,
- * a live lock may happen if the compaction happens too frequently: one processing keeps deleting
- * old files while another one keeps retrying. Setting a reasonable cleanup delay could avoid it.
- */
- private val fileCleanupDelayMs = sparkSession.sessionState.conf.fileSinkLogCleanupDelay
+ protected override val fileCleanupDelayMs =
--- End diff --
Oh, sorry about it, will fix it now.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #13513: [SPARK-15698][SQL][Streaming] Add the ability to remove ...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on the issue:
https://github.com/apache/spark/pull/13513
Sorry. Replied a wrong PR. Deleting.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #13513: [SPARK-15698][SQL][Streaming] Add the ability to ...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/13513#discussion_r79023964
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSourceLog.scala ---
@@ -0,0 +1,133 @@
+/*
+ * 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.execution.streaming
+
+import scala.collection.mutable
+
+import org.json4s.NoTypeHints
+import org.json4s.jackson.Serialization
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.execution.streaming.FileStreamSource.FileEntry
+import org.apache.spark.sql.internal.SQLConf
+
+class FileStreamSourceLog(
+ metadataLogVersion: String,
+ sparkSession: SparkSession,
+ path: String)
+ extends CompactibleFileStreamLog[FileEntry](metadataLogVersion, sparkSession, path) {
+
+ import CompactibleFileStreamLog._
+
+ // Configurations about metadata compaction
+ protected override val compactInterval =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL)
+ require(compactInterval > 0,
+ s"Please set ${SQLConf.FILE_SOURCE_LOG_COMPACT_INTERVAL.key} (was $compactInterval) to a " +
+ s"positive value.")
+
+ protected override val fileCleanupDelayMs =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_CLEANUP_DELAY)
+
+ protected override val isDeletingExpiredLog =
+ sparkSession.conf.get(SQLConf.FILE_SOURCE_LOG_DELETION)
+
+ private implicit val formats = Serialization.formats(NoTypeHints)
+
+ // A fixed size log cache to cache the file entries belong to the compaction batch. It is used
+ // to avoid scanning the compacted log file to retrieve it's own batch data.
+ private val cacheSize = compactInterval
+ private val fileEntryCache = new mutable.LinkedHashMap[Long, Array[FileEntry]]
+
+ private def updateCache(batchId: Long, logs: Array[FileEntry]): Unit = {
+ if (fileEntryCache.size >= cacheSize) {
+ fileEntryCache.drop(1)
--- End diff --
`drop` doesn't change the original map.
```Scala
scala> val m = scala.collection.mutable.LinkedHashMap[Int, Int]()
m: scala.collection.mutable.LinkedHashMap[Int,Int] = Map()
scala>
scala> m(2) = 1
scala> m
res1: scala.collection.mutable.LinkedHashMap[Int,Int] = Map(2 -> 1)
scala> m.drop(1)
res2: scala.collection.mutable.LinkedHashMap[Int,Int] = Map()
scala> m
res3: scala.collection.mutable.LinkedHashMap[Int,Int] = Map(2 -> 1)
```
I think it should be Java LinkedHashMap. This is an example: https://github.com/apache/spark/blob/03d46aafe561b03e25f4e25cf01e631c18dd827c/streaming/src/main/scala/org/apache/spark/streaming/ui/StreamingJobProgressListener.scala#L45
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org