You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by petermaxlee <gi...@git.apache.org> on 2016/08/20 06:40:14 UTC
[GitHub] spark pull request #14728: [SPARK-17165][SQL] FileStreamSource should not tr...
GitHub user petermaxlee opened a pull request:
https://github.com/apache/spark/pull/14728
[SPARK-17165][SQL] FileStreamSource should not track the list of seen files indefinitely
## What changes were proposed in this pull request?
Before this change, FileStreamSource uses an in-memory hash set to track the list of files processed by the engine. The list can grow indefinitely, leading to OOM or overflow of the hash set.
This patch introduces a new user-defined option called "maxFileAge", default to 24 hours. If a file is older than this age, FileStreamSource will purge it from the in-memory map that was used to track the list of files that have been processed.
## How was this patch tested?
Added unit tests for the underlying utility, and also added an end-to-end test to validate the purge in FileStreamSourceSuite.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/petermaxlee/spark SPARK-17165
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/14728.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 #14728
----
commit ce1dd9c4d6a880751e7b3692f9db8597a88f2d05
Author: petermaxlee <pe...@gmail.com>
Date: 2016-08-20T06:35:57Z
[SPARK-17165][SQL] FileStreamSource should not track the list of seen files indefinitely
----
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not tr...
Posted by petermaxlee <gi...@git.apache.org>.
Github user petermaxlee commented on a diff in the pull request:
https://github.com/apache/spark/pull/14728#discussion_r76096158
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala ---
@@ -41,36 +40,59 @@ class FileStreamSource(
metadataPath: String,
options: Map[String, String]) extends Source with Logging {
+ import FileStreamSource._
+
+ private val sourceOptions = new FileStreamOptions(options)
+
private val fs = new Path(path).getFileSystem(sparkSession.sessionState.newHadoopConf())
private val qualifiedBasePath = fs.makeQualified(new Path(path)) // can contains glob patterns
- private val metadataLog = new HDFSMetadataLog[Seq[String]](sparkSession, metadataPath)
+
+ private val metadataLog = new HDFSMetadataLog[Seq[FileEntry]](sparkSession, metadataPath)
+
private var maxBatchId = metadataLog.getLatest().map(_._1).getOrElse(-1L)
/** Maximum number of new files to be considered in each batch */
- private val maxFilesPerBatch = getMaxFilesPerBatch()
+ private val maxFilesPerBatch = sourceOptions.maxFilesPerTrigger
+
+ /** A mapping from a file that we have processed to some timestamp it was last modified. */
+ // Visible for testing.
+ val seenFiles = new SeenFilesMap(sourceOptions.maxFileAgeMs)
--- End diff --
This is actually useful in general for debugging. I'm going to update the comment.
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not tr...
Posted by petermaxlee <gi...@git.apache.org>.
Github user petermaxlee commented on a diff in the pull request:
https://github.com/apache/spark/pull/14728#discussion_r76095002
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala ---
@@ -41,36 +40,59 @@ class FileStreamSource(
metadataPath: String,
options: Map[String, String]) extends Source with Logging {
+ import FileStreamSource._
+
+ private val sourceOptions = new FileStreamOptions(options)
+
private val fs = new Path(path).getFileSystem(sparkSession.sessionState.newHadoopConf())
private val qualifiedBasePath = fs.makeQualified(new Path(path)) // can contains glob patterns
- private val metadataLog = new HDFSMetadataLog[Seq[String]](sparkSession, metadataPath)
+
+ private val metadataLog = new HDFSMetadataLog[Seq[FileEntry]](sparkSession, metadataPath)
+
private var maxBatchId = metadataLog.getLatest().map(_._1).getOrElse(-1L)
/** Maximum number of new files to be considered in each batch */
- private val maxFilesPerBatch = getMaxFilesPerBatch()
+ private val maxFilesPerBatch = sourceOptions.maxFilesPerTrigger
+
+ /** A mapping from a file that we have processed to some timestamp it was last modified. */
+ // Visible for testing.
+ val seenFiles = new SeenFilesMap(sourceOptions.maxFileAgeMs)
- private val seenFiles = new OpenHashSet[String]
- metadataLog.get(None, Some(maxBatchId)).foreach { case (batchId, files) =>
- files.foreach(seenFiles.add)
+ metadataLog.get(None, Some(maxBatchId)).foreach { case (batchId, entry) =>
+ entry.foreach(seenFiles.add)
+ seenFiles.purge()
--- End diff --
This is not safe until we truncate the logs.
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14728
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64260/
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 #14728: [SPARK-17165][SQL] FileStreamSource should not tr...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on a diff in the pull request:
https://github.com/apache/spark/pull/14728#discussion_r75573941
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamOptions.scala ---
@@ -0,0 +1,56 @@
+/*
+ * 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.util.Try
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.execution.datasources.CaseInsensitiveMap
+import org.apache.spark.util.Utils
+
+/**
+ * User specified options for file streams.
+ */
+class FileStreamOptions(@transient private val parameters: Map[String, String])
+ extends Logging with Serializable {
+
+ val maxFilesPerTrigger: Option[Int] = parameters.get("maxFilesPerTrigger").map { str =>
+ Try(str.toInt).toOption.filter(_ > 0).getOrElse {
+ throw new IllegalArgumentException(
+ s"Invalid value '$str' for option 'maxFilesPerTrigger', must be a positive integer")
+ }
+ }
+
+ /** Maximum age of a file that can be found in this directory, before it is deleted. */
+ val maxFileAgeMs: Long =
+ Utils.timeStringAsMs(parameters.getOrElse("maxFileAge", "24h"))
--- End diff --
24 hour seems too short. Maybe a month?
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not tr...
Posted by petermaxlee <gi...@git.apache.org>.
Github user petermaxlee commented on a diff in the pull request:
https://github.com/apache/spark/pull/14728#discussion_r75573771
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamOptions.scala ---
@@ -0,0 +1,56 @@
+/*
+ * 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.util.Try
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.execution.datasources.CaseInsensitiveMap
+import org.apache.spark.util.Utils
+
+/**
+ * User specified options for file streams.
+ */
+class FileStreamOptions(@transient private val parameters: Map[String, String])
--- End diff --
This is a similar setup to CSVOptions and JSONOptions. I felt it would be easier to track the list of options read by the source here.
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14728
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64133/
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on the issue:
https://github.com/apache/spark/pull/14728
Looks pretty good. Just one comment about `Serializable`.
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14728
**[Test build #64362 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64362/consoleFull)** for PR 14728 at commit [`04a112d`](https://github.com/apache/spark/commit/04a112dda448c9ac610626d814ae1d9c3e6185d8).
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14728
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on the issue:
https://github.com/apache/spark/pull/14728
For the definition of `maxAge`, currently from the code it is max age to latest file, people may misunderstand it is max age to current time, so it would be better to document the meaning.
Another concern is that this patch changes the `MetadataLog` format, which will break upgrade compatibility, also user with old metadata log existed will encounter exception when using new 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 issue #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/14728
cc @tdas
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14728
**[Test build #64454 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64454/consoleFull)** for PR 14728 at commit [`9a5ed19`](https://github.com/apache/spark/commit/9a5ed19f3b397b991794a6852aebb2b14c83d635).
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not tr...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/14728#discussion_r76096113
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala ---
@@ -41,36 +40,59 @@ class FileStreamSource(
metadataPath: String,
options: Map[String, String]) extends Source with Logging {
+ import FileStreamSource._
+
+ private val sourceOptions = new FileStreamOptions(options)
+
private val fs = new Path(path).getFileSystem(sparkSession.sessionState.newHadoopConf())
private val qualifiedBasePath = fs.makeQualified(new Path(path)) // can contains glob patterns
- private val metadataLog = new HDFSMetadataLog[Seq[String]](sparkSession, metadataPath)
+
+ private val metadataLog = new HDFSMetadataLog[Seq[FileEntry]](sparkSession, metadataPath)
+
--- End diff --
I know it's bike-shedding, but, my personal taste is to keep declaration of member fields together into a block, or if there are many, perhaps logically group them. So just really depends on whether you mean metadataLog to stand a bit separately from other fields.
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14728
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64336/
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on the issue:
https://github.com/apache/spark/pull/14728
LGTM. Thanks! Merging into 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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14728
**[Test build #64362 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64362/consoleFull)** for PR 14728 at commit [`04a112d`](https://github.com/apache/spark/commit/04a112dda448c9ac610626d814ae1d9c3e6185d8).
* 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 pull request #14728: [SPARK-17165][SQL] FileStreamSource should not tr...
Posted by petermaxlee <gi...@git.apache.org>.
Github user petermaxlee commented on a diff in the pull request:
https://github.com/apache/spark/pull/14728#discussion_r76095989
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala ---
@@ -41,36 +40,59 @@ class FileStreamSource(
metadataPath: String,
options: Map[String, String]) extends Source with Logging {
+ import FileStreamSource._
+
+ private val sourceOptions = new FileStreamOptions(options)
+
private val fs = new Path(path).getFileSystem(sparkSession.sessionState.newHadoopConf())
private val qualifiedBasePath = fs.makeQualified(new Path(path)) // can contains glob patterns
- private val metadataLog = new HDFSMetadataLog[Seq[String]](sparkSession, metadataPath)
+
+ private val metadataLog = new HDFSMetadataLog[Seq[FileEntry]](sparkSession, metadataPath)
+
private var maxBatchId = metadataLog.getLatest().map(_._1).getOrElse(-1L)
/** Maximum number of new files to be considered in each batch */
- private val maxFilesPerBatch = getMaxFilesPerBatch()
+ private val maxFilesPerBatch = sourceOptions.maxFilesPerTrigger
+
+ /** A mapping from a file that we have processed to some timestamp it was last modified. */
+ // Visible for testing.
+ val seenFiles = new SeenFilesMap(sourceOptions.maxFileAgeMs)
- private val seenFiles = new OpenHashSet[String]
- metadataLog.get(None, Some(maxBatchId)).foreach { case (batchId, files) =>
- files.foreach(seenFiles.add)
+ metadataLog.get(None, Some(maxBatchId)).foreach { case (batchId, entry) =>
+ entry.foreach(seenFiles.add)
+ seenFiles.purge()
--- End diff --
good idea. let me add it to the option class.
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14728
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64362/
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by petermaxlee <gi...@git.apache.org>.
Github user petermaxlee commented on the issue:
https://github.com/apache/spark/pull/14728
Re: format compatibility
My understanding is that structured streaming is not yet production ready and the compatibility guarantee across versions doesn't really apply. Otherwise, we shouldn't use Java serialization here at all and should opt for explicit JSON or protobuf.
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by petermaxlee <gi...@git.apache.org>.
Github user petermaxlee commented on the issue:
https://github.com/apache/spark/pull/14728
I've updated the default and set it to 1 week.
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14728
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14728
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14728
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64454/
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the issue:
https://github.com/apache/spark/pull/14728
@jerryshao want to review this?
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on the issue:
https://github.com/apache/spark/pull/14728
Sure, let me take a look at this.
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14728
**[Test build #64133 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64133/consoleFull)** for PR 14728 at commit [`ce1dd9c`](https://github.com/apache/spark/commit/ce1dd9c4d6a880751e7b3692f9db8597a88f2d05).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* ` case class FileEntry(path: String, timestamp: Timestamp) extends Serializable`
* ` class SeenFilesMap(maxAgeMs: Long) `
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not tr...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on a diff in the pull request:
https://github.com/apache/spark/pull/14728#discussion_r76011328
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala ---
@@ -41,36 +40,59 @@ class FileStreamSource(
metadataPath: String,
options: Map[String, String]) extends Source with Logging {
+ import FileStreamSource._
+
+ private val sourceOptions = new FileStreamOptions(options)
+
private val fs = new Path(path).getFileSystem(sparkSession.sessionState.newHadoopConf())
private val qualifiedBasePath = fs.makeQualified(new Path(path)) // can contains glob patterns
- private val metadataLog = new HDFSMetadataLog[Seq[String]](sparkSession, metadataPath)
+
+ private val metadataLog = new HDFSMetadataLog[Seq[FileEntry]](sparkSession, metadataPath)
+
private var maxBatchId = metadataLog.getLatest().map(_._1).getOrElse(-1L)
/** Maximum number of new files to be considered in each batch */
- private val maxFilesPerBatch = getMaxFilesPerBatch()
+ private val maxFilesPerBatch = sourceOptions.maxFilesPerTrigger
+
+ /** A mapping from a file that we have processed to some timestamp it was last modified. */
+ // Visible for testing.
+ val seenFiles = new SeenFilesMap(sourceOptions.maxFileAgeMs)
- private val seenFiles = new OpenHashSet[String]
- metadataLog.get(None, Some(maxBatchId)).foreach { case (batchId, files) =>
- files.foreach(seenFiles.add)
+ metadataLog.get(None, Some(maxBatchId)).foreach { case (batchId, entry) =>
+ entry.foreach(seenFiles.add)
+ seenFiles.purge()
--- End diff --
From my understanding, this `purge()` can be moved out of this loop and do only once, since it is time consuming.
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not tr...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/14728
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14728
**[Test build #64133 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64133/consoleFull)** for PR 14728 at commit [`ce1dd9c`](https://github.com/apache/spark/commit/ce1dd9c4d6a880751e7b3692f9db8597a88f2d05).
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14728
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 pull request #14728: [SPARK-17165][SQL] FileStreamSource should not tr...
Posted by petermaxlee <gi...@git.apache.org>.
Github user petermaxlee commented on a diff in the pull request:
https://github.com/apache/spark/pull/14728#discussion_r76095443
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala ---
@@ -41,36 +40,59 @@ class FileStreamSource(
metadataPath: String,
options: Map[String, String]) extends Source with Logging {
+ import FileStreamSource._
+
+ private val sourceOptions = new FileStreamOptions(options)
+
private val fs = new Path(path).getFileSystem(sparkSession.sessionState.newHadoopConf())
private val qualifiedBasePath = fs.makeQualified(new Path(path)) // can contains glob patterns
- private val metadataLog = new HDFSMetadataLog[Seq[String]](sparkSession, metadataPath)
+
+ private val metadataLog = new HDFSMetadataLog[Seq[FileEntry]](sparkSession, metadataPath)
+
--- End diff --
There is only one blank line. It's actually a good idea to have blank lines separating variable definitions, as documented in a lot of different coding style guides. Excessive blank lines are bad though.
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14728
**[Test build #64260 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64260/consoleFull)** for PR 14728 at commit [`a371f05`](https://github.com/apache/spark/commit/a371f05843d1eae3ae09e21bbd1cedeffb19d0e2).
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by petermaxlee <gi...@git.apache.org>.
Github user petermaxlee commented on the issue:
https://github.com/apache/spark/pull/14728
cc @rxin @cloud-fan
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not tr...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on a diff in the pull request:
https://github.com/apache/spark/pull/14728#discussion_r76011569
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala ---
@@ -41,36 +40,59 @@ class FileStreamSource(
metadataPath: String,
options: Map[String, String]) extends Source with Logging {
+ import FileStreamSource._
+
+ private val sourceOptions = new FileStreamOptions(options)
+
private val fs = new Path(path).getFileSystem(sparkSession.sessionState.newHadoopConf())
private val qualifiedBasePath = fs.makeQualified(new Path(path)) // can contains glob patterns
- private val metadataLog = new HDFSMetadataLog[Seq[String]](sparkSession, metadataPath)
+
+ private val metadataLog = new HDFSMetadataLog[Seq[FileEntry]](sparkSession, metadataPath)
+
private var maxBatchId = metadataLog.getLatest().map(_._1).getOrElse(-1L)
/** Maximum number of new files to be considered in each batch */
- private val maxFilesPerBatch = getMaxFilesPerBatch()
+ private val maxFilesPerBatch = sourceOptions.maxFilesPerTrigger
+
+ /** A mapping from a file that we have processed to some timestamp it was last modified. */
+ // Visible for testing.
+ val seenFiles = new SeenFilesMap(sourceOptions.maxFileAgeMs)
--- End diff --
Actually you could use scalatest `PrivateMethodTester` to test private method, it is not necessary to expose only for testing.
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not tr...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/14728#discussion_r75786696
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala ---
@@ -17,21 +17,18 @@
package org.apache.spark.sql.execution.streaming
-import scala.util.Try
+import scala.collection.JavaConverters._
import org.apache.hadoop.fs.Path
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
-import org.apache.spark.sql.execution.datasources.{CaseInsensitiveMap, DataSource, ListingFileCatalog, LogicalRelation}
+import org.apache.spark.sql.execution.datasources.{DataSource, ListingFileCatalog, LogicalRelation}
import org.apache.spark.sql.types.StructType
-import org.apache.spark.util.collection.OpenHashSet
/**
- * A very simple source that reads text files from the given directory as they appear.
- *
- * TODO Clean up the metadata files periodically
--- End diff --
This TODO still applies right?
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/14728
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 pull request #14728: [SPARK-17165][SQL] FileStreamSource should not tr...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/14728#discussion_r76339023
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamOptions.scala ---
@@ -0,0 +1,56 @@
+/*
+ * 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.util.Try
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.execution.datasources.CaseInsensitiveMap
+import org.apache.spark.util.Utils
+
+/**
+ * User specified options for file streams.
+ */
+class FileStreamOptions(@transient private val parameters: Map[String, String])
--- End diff --
You can remove `Serializable` and `@transient` from this class. It's not used in the executor side.
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not tr...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on a diff in the pull request:
https://github.com/apache/spark/pull/14728#discussion_r76011102
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamOptions.scala ---
@@ -0,0 +1,59 @@
+/*
+ * 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.util.Try
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.execution.datasources.CaseInsensitiveMap
+import org.apache.spark.util.Utils
+
+/**
+ * User specified options for file streams.
+ */
+class FileStreamOptions(@transient private val parameters: Map[String, String])
+ extends Logging with Serializable {
+
+ val maxFilesPerTrigger: Option[Int] = parameters.get("maxFilesPerTrigger").map { str =>
+ Try(str.toInt).toOption.filter(_ > 0).getOrElse {
+ throw new IllegalArgumentException(
+ s"Invalid value '$str' for option 'maxFilesPerTrigger', must be a positive integer")
+ }
+ }
+
+ /**
+ * Maximum age of a file that can be found in this directory, before it is deleted.
+ * Default to a week.
+ */
+ val maxFileAgeMs: Long =
+ Utils.timeStringAsMs(parameters.getOrElse("maxFileAge", "7d"))
+
+ /** Options as specified by the user, in a case-insensitive map, without "path" set. */
+ val optionMapWithoutPath: Map[String, String] =
+ new CaseInsensitiveMap(parameters).filterKeys(_ != "path")
+}
+
+
+object FileStreamOptions {
+
+ def apply(): FileStreamOptions = new FileStreamOptions(Map.empty)
+
+ def apply(paramName: String, paramValue: String): FileStreamOptions = {
+ new FileStreamOptions(Map(paramName -> paramValue))
+ }
--- End diff --
Looks like these two `apply` methods are not used, it would be better to remove them if not used.
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14728
**[Test build #64336 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64336/consoleFull)** for PR 14728 at commit [`6817dec`](https://github.com/apache/spark/commit/6817dec737c4a9e8e4007a92f8f65a1fbc89869e).
* 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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14728
**[Test build #64454 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64454/consoleFull)** for PR 14728 at commit [`9a5ed19`](https://github.com/apache/spark/commit/9a5ed19f3b397b991794a6852aebb2b14c83d635).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `class FileStreamOptions(parameters: Map[String, String]) extends Logging `
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not tr...
Posted by jerryshao <gi...@git.apache.org>.
Github user jerryshao commented on a diff in the pull request:
https://github.com/apache/spark/pull/14728#discussion_r76015528
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala ---
@@ -41,36 +40,59 @@ class FileStreamSource(
metadataPath: String,
options: Map[String, String]) extends Source with Logging {
+ import FileStreamSource._
+
+ private val sourceOptions = new FileStreamOptions(options)
+
private val fs = new Path(path).getFileSystem(sparkSession.sessionState.newHadoopConf())
private val qualifiedBasePath = fs.makeQualified(new Path(path)) // can contains glob patterns
- private val metadataLog = new HDFSMetadataLog[Seq[String]](sparkSession, metadataPath)
+
+ private val metadataLog = new HDFSMetadataLog[Seq[FileEntry]](sparkSession, metadataPath)
+
--- End diff --
Remove unnecessary blank lines, that will be more organized.
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not tr...
Posted by petermaxlee <gi...@git.apache.org>.
Github user petermaxlee commented on a diff in the pull request:
https://github.com/apache/spark/pull/14728#discussion_r75795996
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala ---
@@ -17,21 +17,18 @@
package org.apache.spark.sql.execution.streaming
-import scala.util.Try
+import scala.collection.JavaConverters._
import org.apache.hadoop.fs.Path
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.internal.Logging
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
-import org.apache.spark.sql.execution.datasources.{CaseInsensitiveMap, DataSource, ListingFileCatalog, LogicalRelation}
+import org.apache.spark.sql.execution.datasources.{DataSource, ListingFileCatalog, LogicalRelation}
import org.apache.spark.sql.types.StructType
-import org.apache.spark.util.collection.OpenHashSet
/**
- * A very simple source that reads text files from the given directory as they appear.
- *
- * TODO Clean up the metadata files periodically
--- End diff --
Put it back. Also updated the file with a new test case to make the seen map more robust.
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14728
**[Test build #64336 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64336/consoleFull)** for PR 14728 at commit [`6817dec`](https://github.com/apache/spark/commit/6817dec737c4a9e8e4007a92f8f65a1fbc89869e).
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not tr...
Posted by petermaxlee <gi...@git.apache.org>.
Github user petermaxlee commented on a diff in the pull request:
https://github.com/apache/spark/pull/14728#discussion_r76108686
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala ---
@@ -41,36 +40,59 @@ class FileStreamSource(
metadataPath: String,
options: Map[String, String]) extends Source with Logging {
+ import FileStreamSource._
+
+ private val sourceOptions = new FileStreamOptions(options)
+
private val fs = new Path(path).getFileSystem(sparkSession.sessionState.newHadoopConf())
private val qualifiedBasePath = fs.makeQualified(new Path(path)) // can contains glob patterns
- private val metadataLog = new HDFSMetadataLog[Seq[String]](sparkSession, metadataPath)
+
+ private val metadataLog = new HDFSMetadataLog[Seq[FileEntry]](sparkSession, metadataPath)
+
--- End diff --
metadataLog is not logically related to qualifiedBasePath
---
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 #14728: [SPARK-17165][SQL] FileStreamSource should not track the...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/14728
**[Test build #64260 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64260/consoleFull)** for PR 14728 at commit [`a371f05`](https://github.com/apache/spark/commit/a371f05843d1eae3ae09e21bbd1cedeffb19d0e2).
* 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