You are viewing a plain text version of this content. The canonical link for it is here.

Posted to reviews@spark.apache.org by steveloughran <gi...@git.apache.org> on 2016/08/20 11:18:38 UTC

[GitHub] spark pull request #14731: [SPARK-17159] [streaming]: optimise check for new...

GitHub user steveloughran opened a pull request:

    https://github.com/apache/spark/pull/14731

    [SPARK-17159] [streaming]: optimise check for new files in FileInputDStream

    ## What changes were proposed in this pull request?
    
    This PR optimises the filesystem metadata reads in `FileInputDStream`, by moving the filters used in `FileSystem.globStatus` and `FileSystem.listStatus` into filtering of the `FileStatus` instances returned in the results, so avoiding the need to create `FileStatus` instances within the `FileSystem` operation.
    
    * This doesn't add overhead to the filtering process; that's done as post-processing in the`FileSystem` glob/list operations anyway.
    * At worst it may result in larger lists being built up and returned.
    * For every glob match of a file, the code saves 1 RPC calls to the HDFS NN; 1 GET against S3
    * For every glob match of a directory, the code the code saves 1 RPC call and 2-3 HTTP calls to S3 for the directory check (including a slow List call whenever the directory has children as it doesn't exist as a blob any more)
    * for the modtime check of every file, it saves a Hadoop RPC call, against all object stores *which don't implement any client-side cache*, an HTTP GET. 
    * By entirely eliminating all `getFileStatus()` calls in the listed files, it should reduce the risk of AWS S3 throttling the HTTP request, as it does when too many requests are made to parts of a single S3 bucket.
    
    ## How was this patch tested?
    
    Running the spark streaming tests as a regression suite. In the SPARK-7481 cloud code, I could add a test against S3 which prints to stdout the exact number of HTTP requests made to S3 before and after the patch, so as to validate speedup. (the S3A metrics in Hadoop 2.8+ are accessible at the API level, but as they are only accessible in a new API added in 2.8; it'd stop that proposed module building against Hadoop 2.7. Logging and manual assessment is the only cross-version strategy.

You can merge this pull request into a Git repository by running:

    $ git pull https://github.com/steveloughran/spark cloud/SPARK-17159-listfiles

Alternatively you can review and apply these changes as the patch at:

    https://github.com/apache/spark/pull/14731.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 #14731
    
----
commit 738c51bb57f331c58a877aa20aa5e2beb1084114
Author: Steve Loughran <st...@apache.org>
Date:   2016-08-20T10:50:34Z

    SPARK-17159: move filtering of directories and files out of glob/list filters and into filtering of the FileStatus instances returned in the results, so avoiding the need to create FileStatus intances for
    
    -This doesn't add overhead to the filtering process; that's done as post-processing in FileSystem anyway. At worst it may result in larger lists being built up and returned.
    -For every glob match, the code saves 2 RPC calls to the HDFS NN
    -The code saves 1-3 HTTP calls to S3 for the directory check (including a slow List call whenever the directory has children as it doesn't exist as a blob any more)
    -for the modtime check of every file, it saves an HTTP GET
    
    The whole modtime cache can be eliminated; it's a performance optimisation to avoid the overhead of the file checks, one that is no longer needed.

----


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #73433 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/73433/testReport)** for PR 14731 at commit [`04f4967`](https://github.com/apache/spark/commit/04f49679b3f4f3e2d99e7cafeb9e4fa91fe98ece).


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r75944182
  
    --- Diff: streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala ---
    @@ -196,29 +192,33 @@ class FileInputDStream[K, V, F <: NewInputFormat[K, V]](
           logDebug(s"Getting new files for time $currentTime, " +
             s"ignoring files older than $modTimeIgnoreThreshold")
     
    -      val newFileFilter = new PathFilter {
    -        def accept(path: Path): Boolean = isNewFile(path, currentTime, modTimeIgnoreThreshold)
    -      }
    -      val directoryFilter = new PathFilter {
    -        override def accept(path: Path): Boolean = fs.getFileStatus(path).isDirectory
    +      val directories = if (SparkHadoopUtil.get.isGlobPath(directoryPath)) {
    +        fs.globStatus(directoryPath)
    +            .filter(_.isDirectory)
    +            .map(_.getPath)
    +      } else {
    +        List(directoryPath).toArray
    --- End diff --
    
    Just `Array(directoryPath)`?


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Ok. what is the way? Do I write a formal proposal?
    
    Because right now there is no reliable way to get the full dependency graph of Spark + hadoop cloud JARs + direct cloud provider JARs (azure,aws) and their dependencies (jackson) in sync. 
    
    Which means that getting Spark to talk to object stores is more miss than hit.
    
    I'm happy to follow the proposal mechanism, including progress reports &c, but I do at least need some kind of hope that my work will actually get in.


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Test PASSed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64142/
    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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r107194624
  
    --- Diff: streaming/src/test/scala/org/apache/spark/streaming/InputStreamsSuite.scala ---
    @@ -27,7 +27,8 @@ import scala.collection.JavaConverters._
     import scala.collection.mutable
     
     import com.google.common.io.Files
    -import org.apache.hadoop.fs.Path
    +import org.apache.commons.io.IOUtils
    --- End diff --
    
    Actually, there's a straightforward reason: the test is using the hadoop FS APIs, opening an input stream from a Path and writing to it; `Files.write` is working with a local file. It doesn't work with hadoop FileSystem and Path classes, so could only be used by abusing knowledge of path URLs. Going through FileSystem/Path uses the same API as you'd use in production, so is the more rigorous test.


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    The latest patch pulls out the shortcutting of the globStatus call if there's no wildcard chars in the path; closer to the original patch


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Test PASSed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/73433/
    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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

https://github.com/apache/spark/pull/14731#discussion_r76116321

--- Diff: docs/streaming-programming-guide.md ---
@@ -644,13 +644,39 @@ methods for creating DStreams from files as input sources.
</div>
</div>

Oh, is this not a regular expression? I'd change the doc then to not describe it as one. It sounds like it's following some other kind of glob syntax.

---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    The Hadoop FS Spec has now been updated to declare exactly what HDFS does w.r.t timestamps, and warn that what other filesystems and object stores do are implementation and installation specific features: [filesystem.md](https://github.com/apache/hadoop/blob/trunk/hadoop-common-project/hadoop-common/src/site/markdown/filesystem/introduction.md)
    
    That is the associated documentation update with this one; some of the content there was originally here, but moved over to the hadoop docs for the HDFS team to take the blame for when it changes.


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Test PASSed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/73434/
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Test PASSed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/65592/
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #64534 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64534/consoleFull)** for PR 14731 at commit [`4134620`](https://github.com/apache/spark/commit/4134620210e28a2e182397a9bc94ccb8c4d5ffc4).


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r76573274
  
    --- Diff: core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala ---
    @@ -244,6 +244,31 @@ class SparkHadoopUtil extends Logging {
       }
     
       /**
    +   * List directories/files matching the path and return the `FileStatus` results.
    +   * If the pattern is not a regexp then a simple `getFileStatus(pattern)`
    +   * is called to get the status of that path.
    +   * If the path/pattern does not match anything in the filesystem,
    +   * an empty sequence is returned.
    +   * @param pattern pattern
    +   * @return a possibly empty array of FileStatus entries
    +   */
    +  def globToFileStatus(pattern: Path): Array[FileStatus] = {
    --- End diff --
    
    Here I'm narrowly concerned with the ambiguity of the behavior of a single method, because you can't distinguish between a path with a "?" in it and a glob wildcard for example. The rest seems orthogonal?
    
    The change as it stood to resolve the issue in the OP seemed OK. This is bigger now and I'm not as sure about the rest of the change.


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #71866 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/71866/testReport)** for PR 14731 at commit [`06b2bee`](https://github.com/apache/spark/commit/06b2beec75084db1ee330fa4ff4d50775d9f540c).


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r97363419
  
    --- Diff: streaming/src/test/scala/org/apache/spark/streaming/InputStreamsSuite.scala ---
    @@ -235,18 +236,97 @@ class InputStreamsSuite extends TestSuiteBase with BeforeAndAfter {
             }
             // Over time, create files in the temp directory 1
             val input1 = Seq(1, 2, 3, 4, 5)
    -        input1.foreach(i => createFileAndAdvenceTime(i, testSubDir1))
    +        input1.foreach(i => createFileAndAdvanceTime(i, testSubDir1))
     
             // Over time, create files in the temp directory 1
             val input2 = Seq(6, 7, 8, 9, 10)
    -        input2.foreach(i => createFileAndAdvenceTime(i, testSubDir2))
    +        input2.foreach(i => createFileAndAdvanceTime(i, testSubDir2))
     
             // Verify that all the files have been read
             val expectedOutput = (input1 ++ input2).map(_.toString).toSet
             assert(outputQueue.asScala.flatten.toSet === expectedOutput)
           }
         } finally {
    -      if (testDir != null) Utils.deleteRecursively(testDir)
    +      if (testDir != null) {
    +        Utils.deleteRecursively(testDir)
    +      }
    +    }
    +  }
    +
    +  /**
    +   * Tests that renamed directories are included in new batches -but that only files created
    +   * within the batch window are included.
    +   * Uses the Hadoop APIs to verify consistent behavior with the operations used internally.
    +   */
    +  test("renamed directories are scanned") {
    +    val testDir = Utils.createTempDir()
    --- End diff --
    
    That's in `SQLTestUtils`, and not in the streaming suite can pickup without adding spark-sql:test-jar to the test cp. Or: copy & paste in streaming (`org.apache.spark.streaming.TestSuiteBase`), copy & paste somwhere into spark-core:test-jar, which would be a bigger change.
    
    How about I add it to the spark streaming testbase and then move the other tests in InputStreamSuite to it, either in this patch or a successor? That way: cleaner tests
    



---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Test PASSed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64140/
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran closed the pull request at:

    https://github.com/apache/spark/pull/14731


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r107161471
  
    --- Diff: streaming/src/test/scala/org/apache/spark/streaming/TestSuiteBase.scala ---
    @@ -557,4 +557,16 @@ trait TestSuiteBase extends SparkFunSuite with BeforeAndAfter with Logging {
           verifyOutput[W](output.toSeq, expectedOutput, useSet)
         }
       }
    +
    +  /**
    +   * Creates a temporary directory, which is then passed to `f` and will be deleted after `f`
    +   * returns.
    +   * (originally from `SqlTestUtils`.)
    +   * @todo Probably this method should be moved to a more general place
    +   */
    +  protected def withTempDir(f: File => Unit): Unit = {
    --- End diff --
    
    I see, it's only otherwise defined in the SQL test utils class. Well something we could unify one day, maybe not such a big deal now 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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by rxin <gi...@git.apache.org>.

Github user rxin commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Steve I think the main point is you should also respect the time of reviewers. The way most of your pull requests manifest have been suboptimal: they often start with a very early WIP (which is not necessarily a problem), and once in a while (e.g. a month or two) you update it to almost completely change it. The time itself is a problem. It requires a lot of context switching to review your pull requests. In addition, every time you update it it looks like a complete new giant pull request.


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

https://github.com/apache/spark/pull/14731#discussion_r83491041

--- Diff: docs/streaming-programming-guide.md ---
@@ -644,17 +644,90 @@ methods for creating DStreams from files as input sources.
</div>
</div>

- Spark Streaming will monitor the directory `dataDirectory` and process any files created in that directory (files written in nested directories not supported). Note that
+ Spark Streaming will monitor the directory `dataDirectory` and process any files created in that directory.
+
+ ++ The files must have the same data format.
+ + A simple directory can be monitored, such as `hdfs://namenode:8040/logs/`.
+ All files directly such a path will be processed as they are discovered.
+ + A POSIX glob pattern can be supplied, such as
+ `hdfs://namenode:8040/logs/2016-??-31`.
+ Here, the DStream will consist of all files directly under those directories
+ matching the regular expression.
+ That is: it is a pattern of directories, not of files in directories.
+ + All files must be in the same data format.
+ * A file is considered part of a time period based on its modification time
+ \u2014not its creation time.
+ + Files must be created in/moved under the `dataDirectory` directory/directories by
+ an atomic operation. In HDFS and similar filesystems, this can be done *renaming* them
+ into the data directory from another part of the same filesystem.
+ * If a wildcard is used to identify directories, such as `hdfs://namenode:8040/logs/2016-*`,
+ renaming an entire directory to match the path will add the directory to the list of
+ monitored directories. Only the files in the directory whose modification time is
+ within the current window will be included in the stream.
+ + Once processed, changes to a file within the current window will not cause the file to be reread.
+ That is: *updates are ignored*.
+ + The more files under a directory/wildcard pattern, the longer it will take to
+ scan for changes \u2014even if no files have actually changed.
+ + Calling `FileSystem.setTimes()` to fix the timestamp is a way to have the file picked
+ up in a later window, even if its contents have not changed.

- + The files must have the same data format.
- + The files must be created in the `dataDirectory` by atomically *moving* or *renaming* them into
- the data directory.
- + Once moved, the files must not be changed. So if the files are being continuously appended, the new data will not be read.

For simple text files, there is an easier method `streamingContext.textFileStream(dataDirectory)`. And file streams do not require running a receiver, hence does not require allocating cores.

<span class="badge" style="background-color: grey">Python API</span> `fileStream` is not available in the Python API, only `textFileStream` is available.

+
+ Special points for HDFS
--- End diff --

Does this need to start a new section in the markdown?

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r107152263
  
    --- Diff: streaming/src/test/scala/org/apache/spark/streaming/TestSuiteBase.scala ---
    @@ -557,4 +557,16 @@ trait TestSuiteBase extends SparkFunSuite with BeforeAndAfter with Logging {
           verifyOutput[W](output.toSeq, expectedOutput, useSet)
         }
       }
    +
    +  /**
    +   * Creates a temporary directory, which is then passed to `f` and will be deleted after `f`
    +   * returns.
    +   * (originally from `SqlTestUtils`.)
    +   * @todo Probably this method should be moved to a more general place
    +   */
    +  protected def withTempDir(f: File => Unit): Unit = {
    --- End diff --
    
    yes, but in a module that isn't the one where these tests were, so it'd need more dependency logic or pulling it up into a common module, which, if done properly, makes for a big diff


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Ah right, you already have the modification time for free. Sounds good, remove the caching.


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

https://github.com/apache/spark/pull/14731#discussion_r83490975

--- Diff: docs/streaming-programming-guide.md ---
@@ -644,17 +644,90 @@ methods for creating DStreams from files as input sources.
</div>
</div>

I think this comment is still incorrect. A glob is not a regex. If this is just the syntax other Hadoop APIs support, that seems reasonable, but it should be described that way.

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

https://github.com/apache/spark/pull/14731#discussion_r76110009

--- Diff: docs/streaming-programming-guide.md ---
@@ -644,13 +644,39 @@ methods for creating DStreams from files as input sources.
</div>
</div>

Round () or {} brackets? Because the wildcard pattern used in `isGlobPath()` is `"{}[]*?\\"`. Curly only.

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #64142 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64142/consoleFull)** for PR 14731 at commit [`6e8ace0`](https://github.com/apache/spark/commit/6e8ace0444ec9bdebc7c809a08628891f6de5fd0).
     * 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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by uncleGen <gi...@git.apache.org>.

Github user uncleGen commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r103187577
  
    --- Diff: streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala ---
    @@ -140,7 +137,7 @@ class FileInputDStream[K, V, F <: NewInputFormat[K, V]](
        * a union RDD out of them. Note that this maintains the list of files that were processed
        * in the latest modification time in the previous call to this method. This is because the
        * modification time returned by the FileStatus API seems to return times only at the
    -   * granularity of seconds. And new files may have the same modification time as the
    +   * granularity of seconds in HDFS. And new files may have the same modification time as the
        * latest modification time in the previous call to this method yet was not reported in
    --- End diff --
    
    got 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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    The logic has got complex enough it merits unit tests. Pulling into SparkHadoopUtils itself and writing some for the possible: simple, glob matches one , glob matches 1+, glob doesn't match, file not found


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

https://github.com/apache/spark/pull/14731#discussion_r75946809

--- Diff: docs/streaming-programming-guide.md ---
@@ -644,13 +644,39 @@ methods for creating DStreams from files as input sources.
</div>
</div>

`(2015|2016)` would be what means "the string '2015' or '2016'", and `..` would mean "any two characters"

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    to be precise: the caching of file modification times is superfluous. It's there to avoid the cost of executing `getFileStatus()` on previously scanned files. Once you use the FileStatus returned in a listing, you aren't calling `getFileStatus()`, hence: no need to cache


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r95058826
  
    --- Diff: streaming/src/test/scala/org/apache/spark/streaming/InputStreamsSuite.scala ---
    @@ -235,18 +236,97 @@ class InputStreamsSuite extends TestSuiteBase with BeforeAndAfter {
             }
             // Over time, create files in the temp directory 1
             val input1 = Seq(1, 2, 3, 4, 5)
    -        input1.foreach(i => createFileAndAdvenceTime(i, testSubDir1))
    +        input1.foreach(i => createFileAndAdvanceTime(i, testSubDir1))
     
             // Over time, create files in the temp directory 1
             val input2 = Seq(6, 7, 8, 9, 10)
    -        input2.foreach(i => createFileAndAdvenceTime(i, testSubDir2))
    +        input2.foreach(i => createFileAndAdvanceTime(i, testSubDir2))
     
             // Verify that all the files have been read
             val expectedOutput = (input1 ++ input2).map(_.toString).toSet
             assert(outputQueue.asScala.flatten.toSet === expectedOutput)
           }
         } finally {
    -      if (testDir != null) Utils.deleteRecursively(testDir)
    +      if (testDir != null) {
    +        Utils.deleteRecursively(testDir)
    +      }
    +    }
    +  }
    +
    +  /**
    +   * Tests that renamed directories are included in new batches -but that only files created
    +   * within the batch window are included.
    +   * Uses the Hadoop APIs to verify consistent behavior with the operations used internally.
    +   */
    +  test("renamed directories are scanned") {
    +    val testDir = Utils.createTempDir()
    --- End diff --
    
    Can you use `withTempDir { testDir =>`? if so I think several tests could use the same pattern 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 pull request #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r75584026
  
    --- Diff: streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala ---
    @@ -293,8 +290,8 @@ class FileInputDStream[K, V, F <: NewInputFormat[K, V]](
       }
     
       /** Get file mod time from cache or fetch it from the file system */
    -  private def getFileModTime(path: Path) = {
    -    fileToModTime.getOrElseUpdate(path.toString, fs.getFileStatus(path).getModificationTime())
    +  private def getFileModTime(fs: FileStatus) = {
    --- End diff --
    
    yes, I was just being minimal about the changes. Inlining is easy


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #64142 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64142/consoleFull)** for PR 14731 at commit [`6e8ace0`](https://github.com/apache/spark/commit/6e8ace0444ec9bdebc7c809a08628891f6de5fd0).


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #74990 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/74990/testReport)** for PR 14731 at commit [`a3aaf26`](https://github.com/apache/spark/commit/a3aaf267d2ac30c012b4a71b7a80e28a49ff10be).


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r76512357
  
    --- Diff: core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala ---
    @@ -244,6 +244,31 @@ class SparkHadoopUtil extends Logging {
       }
     
       /**
    +   * List directories/files matching the path and return the `FileStatus` results.
    +   * If the pattern is not a regexp then a simple `getFileStatus(pattern)`
    +   * is called to get the status of that path.
    +   * If the path/pattern does not match anything in the filesystem,
    +   * an empty sequence is returned.
    +   * @param pattern pattern
    +   * @return a possibly empty array of FileStatus entries
    +   */
    +  def globToFileStatus(pattern: Path): Array[FileStatus] = {
    --- End diff --
    
    I'm concerned that this method is necessarily ambiguous, because you can't actually distinguish globs from other paths. Is this really needed? that's why the FS API exposes two methods.


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r102977480
  
    --- Diff: docs/streaming-programming-guide.md ---
    @@ -630,35 +630,106 @@ which creates a DStream from text
     data received over a TCP socket connection. Besides sockets, the StreamingContext API provides
     methods for creating DStreams from files as input sources.
     
    -- **File Streams:** For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as:
    +#### File Streams
    +{:.no_toc}
    +
    +For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as
    +via `StreamingContext.fileStream[KeyClass, ValueClass, InputFormatClass]`.
    +
    +File streams do not require running a receiver, hence does not require allocating cores.
     
    -    <div class="codetabs">
    -    <div data-lang="scala" markdown="1">
    -        streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)
    -    </div>
    -    <div data-lang="java" markdown="1">
    -		streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory);
    -    </div>
    -    <div data-lang="python" markdown="1">
    -		streamingContext.textFileStream(dataDirectory)
    -    </div>
    -    </div>
    +For simple text files, the easiest method is `StreamingContext.textFileStream(dataDirectory)`. 
     
    -	Spark Streaming will monitor the directory `dataDirectory` and process any files created in that directory (files written in nested directories not supported). Note that
    +<div class="codetabs">
    +<div data-lang="scala" markdown="1">
     
    -     + The files must have the same data format.
    -     + The files must be created in the `dataDirectory` by atomically *moving* or *renaming* them into
    -     the data directory.
    -     + Once moved, the files must not be changed. So if the files are being continuously appended, the new data will not be read.
    +{% highlight scala %}
    +streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)
    +{% endhighlight %}
    +For text files
    +
    +{% highlight scala %}
    +streamingContext.textFileStream(dataDirectory)
    +{% endhighlight %}
    +</div>
     
    -	For simple text files, there is an easier method `streamingContext.textFileStream(dataDirectory)`. And file streams do not require running a receiver, hence does not require allocating cores.
    +<div data-lang="java" markdown="1">
    +{% highlight java %}
    +streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory);
    +{% endhighlight %}
    +For text files
     
    -	<span class="badge" style="background-color: grey">Python API</span> `fileStream` is not available in the Python API, only	`textFileStream` is	available.
    +{% highlight java %}
    +streamingContext.textFileStream(dataDirectory);
    +{% endhighlight %}
    +</div>
     
    -- **Streams based on Custom Receivers:** DStreams can be created with data streams received through custom receivers. See the [Custom Receiver
    +<div data-lang="python" markdown="1">
    +`fileStream` is not available in the Python API; only `textFileStream` is available.
    +{% highlight python %}
    +streamingContext.textFileStream(dataDirectory)
    +{% endhighlight %}
    +</div>
    +
    +</div>
    +
    +##### How Directories are Monitored
    +{:.no_toc}
    +
    +Spark Streaming will monitor the directory `dataDirectory` and process any files created in that directory.
    +
    +   * A simple directory can be monitored, such as `hdfs://namenode:8040/logs/`.
    +     All files directly under such a path will be processed as they are discovered.
    +   + A [POSIX glob pattern](http://pubs.opengroup.org/onlinepubs/009695399/utilities/xcu_chap02.html#tag_02_13_02) can be supplied, such as
    +     `hdfs://namenode:8040/logs/2017/*`.
    +     Here, the DStream will consist of all files in the directories
    +     matching the pattern.
    +     That is: it is a pattern of directories, not of files in directories.
    +   + All files must be in the same data format.
    +   * A file is considered part of a time period based on its modification time,
    +     not its creation time.
    +   + Once processed, changes to a file within the current window will not cause the file to be reread.
    +     That is: *updates are ignored*.
    +   + The more files under a directory, the longer it will take to
    +     scan for changes \u2014even if no files have been modified.
    --- End diff --
    
    fixed


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #65592 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65592/consoleFull)** for PR 14731 at commit [`57f697d`](https://github.com/apache/spark/commit/57f697dc718e536f512c856b8e6c8239e1133fd5).
     * 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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r76514866
  
    --- Diff: core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala ---
    @@ -244,6 +244,31 @@ class SparkHadoopUtil extends Logging {
       }
     
       /**
    +   * List directories/files matching the path and return the `FileStatus` results.
    +   * If the pattern is not a regexp then a simple `getFileStatus(pattern)`
    +   * is called to get the status of that path.
    +   * If the path/pattern does not match anything in the filesystem,
    +   * an empty sequence is returned.
    +   * @param pattern pattern
    +   * @return a possibly empty array of FileStatus entries
    +   */
    +  def globToFileStatus(pattern: Path): Array[FileStatus] = {
    --- End diff --
    
    It goes with the `globPathIfNecessary` call \u2014you want to rename it to be consistent.
    
    
    Regarding the FS APIs, There's way to many list operations in the FS APIs, each with different flaws.
    
    1. The simple `list(path, filter): Array[FS]` operations don't scale to a directory with hundreds of thousands of files, hence the remote iterator versions
    1. None of them provide any consistency guarantees. Worth knowing. This is more common in remote iterators as the iteration window is bigger, but even in those that return arrays, in a large enough directory things may change during the enum
    1. Anything that treewaks is very suboptimal on blobstores, somewhat inefficient for deep trees.
    1. `listFiles(path, recursive=true)` is the sole one which object stores can currently optimise by avoiding the treewalk and just doing a bulk list. [HADOOP-13208](https://issues.apache.org/jira/browse/HADOOP-13208) has added that for S3A.
    1. ..but that method filters out all directories, which means that apps which do want directories too are out of luck.
    1. globStatus() is even less efficient than the others ... have a look at the source to see why.
    1. In [HADOOP-13371](https://issues.apache.org/jira/browse/HADOOP-13371) I'm exploring an optimised globber, but I don't want to write one which collapses at scale (i.e in production).
    
    I've added some comments in HADOOP-13371 about what to do there, I will probably do that "no regexp -> simple return" strategy implemented in this patch. But it will only benefit s3a in Hadoop 2.8+; patching spark benefits everything.



---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Reynold, I know very much about the time of reviewers, I put 1+h a day on the hadoop codebase reviewing stuff, generally trying to review the work of non-colleagues, so as to pull in the broad set of contributions which are needed.. 
    
    I have been trying to get some object store related patches into spark alongside the foundational work in fundamentally transforming how we work with object storage, especially S3, in Hadoop. Without the spark side changes, a lot gets lost: here the performance is approx 100-300mS/file when scanning an object store. 
    
    here I've split things in two, docs and diff. Both are independent, both are reasonably tractable. If they can be reviewed fast and added, there's no problems of patches ageing, everyone having to resync.
    
    We can get this out the way, and you've have fewer reasons to be unhappy with me.


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

https://github.com/apache/spark/pull/14731#discussion_r94410195

--- Diff: docs/streaming-programming-guide.md ---
@@ -644,17 +644,90 @@ methods for creating DStreams from files as input sources.
</div>
</div>

For simple text files, there is an easier method `streamingContext.textFileStream(dataDirectory)`. And file streams do not require running a receiver, hence does not require allocating cores.

<span class="badge" style="background-color: grey">Python API</span> `fileStream` is not available in the Python API, only `textFileStream` is available.

+
+ Special points for HDFS
+
+ The HDFS filesystem does not update the modification time while it is being written to.
+ Specifically
+
+ + `FileSystem.create()` creation: a zero-byte file is listed; creation and modification time is
+ set to the current time as seen on the NameNode.
+ * Writes to a file via the output stream returned in the `create()` call: the modification
+ time *does not change*.
+ * When `OutputStream.close()` is called, all remaining data is written, the file closed and
+ the NameNode updated with the final size of the file. The modification time is set to
+ the time the file was closed.
+ * File opened for appends via an `append()` operation. This does not change the modification
+ time of the file until the `close()` call is made on the output stream.
+ * `FileSystem.setTimes()` can be used to explicitly set the time on a file.
+ * The rarely used operations: `FileSystem.concat()`, `createSnapshot()`, `createSymlink()` and
+ `truncate()` all update the modification time.
+
+ Together, this means that when a file is opened, even before data has been completely written,
+ it may be included in the DStream -after which updates to the file within the same window
+ will be ignored. That is: changes may be missed, and data omitted from the stream.
+ To guarantee that changes are picked up in a window, write the file
+ to an unmonitored directory, then immediately after the output stream is closed,
+ rename it into the destination directory.
+ Provided the renamed file appears in the scanned destination directory during the window
+ of its creation, the new data will be picked up.
+
+ Object stores have a different set of limitations.
+
+ + Wildcard directory enumeration can very slow, especially if there are many directories
+ or files to scan.
+ * The file only becomes visible at the end of the write operation; this also defines.
+ the creation time of the file.
+ + A file's modification time is always the same as its creation time.
+ + The `FileSystem.setTimes()` command to set file timestamps will be ignored.
+ + `FileSystem.rename(file)` of a single file will update the modification time.
+ The time to rename a file is generally `O(length(file))`: the bigger the file, the longer
+ it takes.
+ + `FileSystem.rename(directory)` is not atomic, and slower the more data there is to rename.
+ It may update the modification times of renamed files.
+ + Object creation directly though a PUT operation is atomic, irrespective of
+ the programming language or library used to upload the data.
+ + Writing a file to an object store using Hadoop's APIs is an atomic operation;
+ the object is only created via a PUT operation in the final `OutputStream.close()` call.
+
+ Applications using an object store as the direct destination of data
+ should use PUT operations to directly publish data for a DStream to pick up \u2014even though this
+ is not the mechanism to use when writing to HDFS or other filesystem. When using the Hadoop
+ FileSystem API in Spark that means: write the data directory directly to the target directory,
+ knowing that it is the final `close()` call will make the file visible and set its creation &
--- End diff --

I've pulled the details on timestamps into the hadoop docs, and cut back to something that says "filesystems: write then rename, object stores: write directly". That is all people really need to know

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    I've now done the [s3a streaming test/example](https://github.com/steveloughran/spark/blob/features/SPARK-7481-cloud/cloud/src/main/scala/org/apache/spark/cloud/s3/examples/S3Streaming.scala)
    
    this uses a pattern of s3a/path/sub* as the directory path; then creates a file in a directory and renames the dir to match the path; verifies that the file was found in the time period allocated
    
    https://gist.github.com/steveloughran/c8b39a7b87a9bd63d7a383bda8687e7e
    
    
    Notable that the scan of the empty dir took 150ms; once there's data in the tree the time jumps up to 500ms once there are two entries under the tree, one dir and one file
    
    summary stats show 72 getFileStatus calls at the FS API, mapping to 140 HEAD calls and 88 LIST operations. 
    
    ```
     S3AFileSystem{uri=s3a://stevel-ireland-new, workingDir=s3a://hwdev-steve-ireland-new/user/stevel, inputPolicy=sequential, partSize=104857600, enableMultiObjectsDelete=true, maxKeys=5000, readAhead=65536, blockSize=1048576, multiPartThreshold=2147483647, statistics {292 bytes read, 292 bytes written, 101 read ops, 0 large read ops, 11 write ops}, metrics {{Context=S3AFileSystem} {FileSystemId=343b706a-c238-4d71-9ed8-8083601ac28a-hwdev-steve-ireland-new} {fsURI=s3a://hwdev-steve-ireland-new} {files_created=1} {files_copied=1} {files_copied_bytes=292} {files_deleted=1} {directories_created=3} {directories_deleted=0} {ignored_errors=2} {op_copy_from_local_file=0} {op_exists=1} {op_get_file_status=72} {op_glob_status=16} {op_is_directory=0} {op_is_file=0} {op_list_files=0} {op_list_located_status=0} {op_list_status=27} {op_mkdirs=2} {op_rename=1} {object_copy_requests=0} {object_delete_requests=3} {object_list_requests=88} {object_continue_list_requests=0} {object_metadata_requests=1
 40} {object_multipart_aborted=0} {object_put_bytes=292} {object_put_requests=4} {stream_read_fully_operations=0} {stream_bytes_skipped_on_seek=0} {stream_bytes_backwards_on_seek=0} {stream_bytes_read=292} {streamOpened=1} {stream_backward_seek_pperations=0} {stream_read_operations_incomplete=0} {stream_bytes_discarded_in_abort=0} {stream_close_operations=1} {stream_read_operations=1} {stream_aborted=0} {stream_forward_seek_operations=0} {streamClosed=1} {stream_seek_operations=0} {stream_bytes_read_in_close=0} {stream_read_exceptions=0} }}
    ```
    
    I'm going to do a test run with the modification here and see what it does to listing and status


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    This is ready to go right @steveloughran ? LGTM


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #64486 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64486/consoleFull)** for PR 14731 at commit [`9bc0ea9`](https://github.com/apache/spark/commit/9bc0ea9734ccaf11c6306a3496c98be8cc20faab).


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r76593850
  
    --- Diff: core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala ---
    @@ -244,6 +244,31 @@ class SparkHadoopUtil extends Logging {
       }
     
       /**
    +   * List directories/files matching the path and return the `FileStatus` results.
    +   * If the pattern is not a regexp then a simple `getFileStatus(pattern)`
    +   * is called to get the status of that path.
    +   * If the path/pattern does not match anything in the filesystem,
    +   * an empty sequence is returned.
    +   * @param pattern pattern
    +   * @return a possibly empty array of FileStatus entries
    +   */
    +  def globToFileStatus(pattern: Path): Array[FileStatus] = {
    --- End diff --
    
    essentially if anything which might be a wildcard is hit, it gets handed off to the globber for the full interpretation. Same for ^ and ], which are only part of a pattern within the context of an opening [
    
    Its only those strings which can be verified to be regexp free in a simple context-free string scan that say "absolutely no patterns here"
    
    regarding the bigger change: most of it is isolation of the sensitive code *and the tests to verify behaviour*


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #71866 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/71866/testReport)** for PR 14731 at commit [`06b2bee`](https://github.com/apache/spark/commit/06b2beec75084db1ee330fa4ff4d50775d9f540c).
     * 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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    @srowen anything else I need to do 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 pull request #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r95058504
  
    --- Diff: docs/streaming-programming-guide.md ---
    @@ -630,35 +630,106 @@ which creates a DStream from text
     data received over a TCP socket connection. Besides sockets, the StreamingContext API provides
     methods for creating DStreams from files as input sources.
     
    -- **File Streams:** For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as:
    +#### File Streams
    +{:.no_toc}
    +
    +For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as
    +via `StreamingContext.fileStream[KeyClass, ValueClass, InputFormatClass]`.
    +
    +File streams do not require running a receiver, hence does not require allocating cores.
     
    -    <div class="codetabs">
    -    <div data-lang="scala" markdown="1">
    -        streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)
    -    </div>
    -    <div data-lang="java" markdown="1">
    -		streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory);
    -    </div>
    -    <div data-lang="python" markdown="1">
    -		streamingContext.textFileStream(dataDirectory)
    -    </div>
    -    </div>
    +For simple text files, the easiest method is `StreamingContext.textFileStream(dataDirectory)`. 
     
    -	Spark Streaming will monitor the directory `dataDirectory` and process any files created in that directory (files written in nested directories not supported). Note that
    +<div class="codetabs">
    +<div data-lang="scala" markdown="1">
     
    -     + The files must have the same data format.
    -     + The files must be created in the `dataDirectory` by atomically *moving* or *renaming* them into
    -     the data directory.
    -     + Once moved, the files must not be changed. So if the files are being continuously appended, the new data will not be read.
    +{% highlight scala %}
    +streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)
    +{% endhighlight %}
    +For text files
    +
    +{% highlight scala %}
    +streamingContext.textFileStream(dataDirectory)
    +{% endhighlight %}
    +</div>
     
    -	For simple text files, there is an easier method `streamingContext.textFileStream(dataDirectory)`. And file streams do not require running a receiver, hence does not require allocating cores.
    +<div data-lang="java" markdown="1">
    +{% highlight java %}
    +streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory);
    +{% endhighlight %}
    +For text files
     
    -	<span class="badge" style="background-color: grey">Python API</span> `fileStream` is not available in the Python API, only	`textFileStream` is	available.
    +{% highlight java %}
    +streamingContext.textFileStream(dataDirectory);
    +{% endhighlight %}
    +</div>
     
    -- **Streams based on Custom Receivers:** DStreams can be created with data streams received through custom receivers. See the [Custom Receiver
    +<div data-lang="python" markdown="1">
    +`fileStream` is not available in the Python API; only `textFileStream` is available.
    +{% highlight python %}
    +streamingContext.textFileStream(dataDirectory)
    +{% endhighlight %}
    +</div>
    +
    +</div>
    +
    +##### How Directories are Monitored
    +{:.no_toc}
    +
    +Spark Streaming will monitor the directory `dataDirectory` and process any files created in that directory.
    +
    +   * A simple directory can be monitored, such as `hdfs://namenode:8040/logs/`.
    +     All files directly under such a path will be processed as they are discovered.
    +   + A [POSIX glob pattern](http://pubs.opengroup.org/onlinepubs/009695399/utilities/xcu_chap02.html#tag_02_13_02) can be supplied, such as
    +     `hdfs://namenode:8040/logs/2017/*`.
    +     Here, the DStream will consist of all files in the directories
    +     matching the pattern.
    +     That is: it is a pattern of directories, not of files in directories.
    +   + All files must be in the same data format.
    +   * A file is considered part of a time period based on its modification time,
    +     not its creation time.
    +   + Once processed, changes to a file within the current window will not cause the file to be reread.
    +     That is: *updates are ignored*.
    +   + The more files under a directory, the longer it will take to
    +     scan for changes \u2014even if no files have been modified.
    --- End diff --
    
    Nits now: space after hyphen if you 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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #66656 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/66656/consoleFull)** for PR 14731 at commit [`f8ed8a3`](https://github.com/apache/spark/commit/f8ed8a3551d1eed5db5a22f5eeb484614036fefe).


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r95079472
  
    --- Diff: docs/streaming-programming-guide.md ---
    @@ -630,35 +630,106 @@ which creates a DStream from text
     data received over a TCP socket connection. Besides sockets, the StreamingContext API provides
     methods for creating DStreams from files as input sources.
     
    -- **File Streams:** For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as:
    +#### File Streams
    +{:.no_toc}
    +
    +For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as
    +via `StreamingContext.fileStream[KeyClass, ValueClass, InputFormatClass]`.
    +
    +File streams do not require running a receiver, hence does not require allocating cores.
     
    -    <div class="codetabs">
    -    <div data-lang="scala" markdown="1">
    -        streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)
    -    </div>
    -    <div data-lang="java" markdown="1">
    -		streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory);
    -    </div>
    -    <div data-lang="python" markdown="1">
    -		streamingContext.textFileStream(dataDirectory)
    -    </div>
    -    </div>
    +For simple text files, the easiest method is `StreamingContext.textFileStream(dataDirectory)`. 
     
    -	Spark Streaming will monitor the directory `dataDirectory` and process any files created in that directory (files written in nested directories not supported). Note that
    +<div class="codetabs">
    +<div data-lang="scala" markdown="1">
     
    -     + The files must have the same data format.
    -     + The files must be created in the `dataDirectory` by atomically *moving* or *renaming* them into
    -     the data directory.
    -     + Once moved, the files must not be changed. So if the files are being continuously appended, the new data will not be read.
    +{% highlight scala %}
    +streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)
    +{% endhighlight %}
    +For text files
    +
    +{% highlight scala %}
    +streamingContext.textFileStream(dataDirectory)
    +{% endhighlight %}
    +</div>
     
    -	For simple text files, there is an easier method `streamingContext.textFileStream(dataDirectory)`. And file streams do not require running a receiver, hence does not require allocating cores.
    +<div data-lang="java" markdown="1">
    +{% highlight java %}
    +streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory);
    +{% endhighlight %}
    +For text files
     
    -	<span class="badge" style="background-color: grey">Python API</span> `fileStream` is not available in the Python API, only	`textFileStream` is	available.
    +{% highlight java %}
    +streamingContext.textFileStream(dataDirectory);
    +{% endhighlight %}
    +</div>
     
    -- **Streams based on Custom Receivers:** DStreams can be created with data streams received through custom receivers. See the [Custom Receiver
    +<div data-lang="python" markdown="1">
    +`fileStream` is not available in the Python API; only `textFileStream` is available.
    +{% highlight python %}
    +streamingContext.textFileStream(dataDirectory)
    +{% endhighlight %}
    +</div>
    +
    +</div>
    +
    +##### How Directories are Monitored
    +{:.no_toc}
    +
    +Spark Streaming will monitor the directory `dataDirectory` and process any files created in that directory.
    +
    +   * A simple directory can be monitored, such as `hdfs://namenode:8040/logs/`.
    +     All files directly under such a path will be processed as they are discovered.
    +   + A [POSIX glob pattern](http://pubs.opengroup.org/onlinepubs/009695399/utilities/xcu_chap02.html#tag_02_13_02) can be supplied, such as
    +     `hdfs://namenode:8040/logs/2017/*`.
    +     Here, the DStream will consist of all files in the directories
    +     matching the pattern.
    +     That is: it is a pattern of directories, not of files in directories.
    +   + All files must be in the same data format.
    +   * A file is considered part of a time period based on its modification time,
    +     not its creation time.
    +   + Once processed, changes to a file within the current window will not cause the file to be reread.
    +     That is: *updates are ignored*.
    +   + The more files under a directory, the longer it will take to
    +     scan for changes \u2014even if no files have been modified.
    +   * If a wildcard is used to identify directories, such as `hdfs://namenode:8040/logs/2016-*`,
    +     renaming an entire directory to match the path will add the directory to the list of
    +     monitored directories. Only the files in the directory whose modification time is
    +     within the current window will be included in the stream.
    +   + Calling `FileSystem.setTimes()` to fix the timestamp is a way to have the file picked
    +     up in a later window, even if its contents have not changed.
    +
    +
    +##### Streaming to FileSystems vs Object stores
    +{:.no_toc}
    +
    +"Full" Filesystems such as HDFS tend to set the modification time on their files as soon
    --- End diff --
    
    FWIW, I've just pulled in some people who have implemented HDFS and S3 clones to give their view of the world, [HADOOP-13946](https://issues.apache.org/jira/browse/HADOOP-13946)
    
    I think for object stores I should actually be vaguer and say "try and it and see". It looks like put ops get their TS when the PUT begins, so multipart put operations will get a TS somewhere between create() and close(). Also, @ajfabbri had to change the semantics on create() to a TS generated by the client when you are using s3guard to cache data.
    
    One troublespot here is that rename() is likely to change timestamps of files in a blobstore, meaning the files window will change from that of creation to that of the rename. That probably is something to call out, as people may need to know of 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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    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 pull request #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r75584030
  
    --- Diff: streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala ---
    @@ -241,16 +233,21 @@ class FileInputDStream[K, V, F <: NewInputFormat[K, V]](
        *   The files with mod time T+5 are not remembered and cannot be ignored (since, t+5 > t+1).
        *   Hence they can get selected as new files again. To prevent this, files whose mod time is more
        *   than current batch time are not considered.
    +   * @param fs file status
    +   * @param currentTime time of the batch
    +   * @param modTimeIgnoreThreshold the ignore threshold
    +   * @return true if the file has been modified within the batch window
        */
    -  private def isNewFile(path: Path, currentTime: Long, modTimeIgnoreThreshold: Long): Boolean = {
    + private def isNewFile(fs: FileStatus, currentTime: Long, modTimeIgnoreThreshold: Long): Boolean = {
    --- End diff --
    
    I'll fix 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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #65498 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65498/consoleFull)** for PR 14731 at commit [`735fc7c`](https://github.com/apache/spark/commit/735fc7c2343c08a323e3d213e611830e3b41ef04).


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Test PASSed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/65498/
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Test FAILed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64486/
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Test PASSed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/71866/
    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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

https://github.com/apache/spark/pull/14731#discussion_r79209668

--- Diff: docs/streaming-programming-guide.md ---
@@ -644,13 +644,44 @@ methods for creating DStreams from files as input sources.
</div>
</div>

- Spark Streaming will monitor the directory `dataDirectory` and process any files created in that directory (files written in nested directories not supported). Note that
-
- + The files must have the same data format.
- + The files must be created in the `dataDirectory` by atomically *moving* or *renaming* them into
- the data directory.
- + Once moved, the files must not be changed. So if the files are being continuously appended, the new data will not be read.
-
+ Spark Streaming will monitor the directory `dataDirectory` and process any files created in that directory.
+
+ + A simple directory can be supplied, such as `hdfs://namenode:8040/logs/`.
+ All files directly such a path will be processed as they are discovered.
+ + A regular expression can be supplied instead, such as
+ `hdfs://namenode:8040/logs/2016-*-31`.
+ Here, the DStream will consist of all files directly under those directories
+ matching the regular expression.
+ That is: it is a pattern of directories, not of files in directories.
+ + All files must be in the same data format.
+ * A file is considered part of a time period based on its modification time
+ \u2014not its creation time.
+ + Files must be created in/moved under the `dataDirectory` directory/directories by
+ an atomic operation. In HDFS and similar filesystems, this can be done *renaming* them
+ into the data directory from another part of the same filesystem.
+ * If a wildcard is used to identify directories, such as `hdfs://namenode:8040/logs/2016*`,
+ renaming an entire directory to match the path will add the directory to the list of
+ monitored directories. However, unless the modification time of the directory's files
+ are within that of the current window, they will not be recognized as new files.
+ + Once processed, changes to a file will not cause the file to be reread.
+ That is: Updates are ignored.
+ + The more files under a directory/wildcard pattern, the longer it will take to
+ scan for changes \u2014even if no files have actually changed.
+
+ Special points for object stores
--- End diff --

For object stores, direct writes to the directory, resulting in a PUT on close(), will guarantee that a file is picked up immediately. Things are actually a bit quirky for HDFS; even file length doesn't get updated reliably during a write-in-progress. I'll add a section there and then ask people who understand HDFS what is really happening

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #64662 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64662/consoleFull)** for PR 14731 at commit [`b60f175`](https://github.com/apache/spark/commit/b60f175b5ef058ed24b3ddaf9a85b899a5e33187).
     * 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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

https://github.com/apache/spark/pull/14731#discussion_r94407115

--- Diff: docs/streaming-programming-guide.md ---
@@ -644,17 +644,90 @@ methods for creating DStreams from files as input sources.
</div>
</div>

I added a link to the posix docs. If you follow them, you eventually end up on some coverage of regexps inside []; the Hadoop Glob code does actually convert the shell expression to a java regexp, then compile it in, so presumably should handle everything that the regexp engine (originally {{java.util.regexp}}, currently {{com.google.re2j}} can compile. That's too much detail and something that should really be covered in the Hadoop docs by someone.

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Test PASSed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64662/
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by uncleGen <gi...@git.apache.org>.

Github user uncleGen commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    @steveloughran Are you still working on 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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Having looked at the source code, `FileSystem.globStatus()` uses the glob patterns, which are not the same as the posix regexp ones. [org.apache.hadoop.fs.GlobPattern](http://grepcode.com/file/repo1.maven.org/maven2/org.apache.hadoop/hadoop-common/2.7.1/org/apache/hadoop/fs/GlobPattern.java#81) does the conversion.
    
    For the docs, I'll just use a wildcard * in the example, rather than try anything more sophisticated.


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    LGTM. I was trying to see if there was a way to create a good test here by triggering the takes-too-long codepath and having a counter, but there's no obvious way to do that deterministically. I am doing a test for this against s3 in the spark-cloud module I'm writing; I can look at the printed counts of getFileStatus before/after the patch to see the difference, but the actual (testable) metrics are only accessible with forthcoming Hadoop 2.8 release.
    
    TL;DR: no easy test, so there's nothing left to do


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #70819 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70819/testReport)** for PR 14731 at commit [`a9a6f7b`](https://github.com/apache/spark/commit/a9a6f7b9e3876e551a2568b6220559992db40228).
     * 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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #64368 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64368/consoleFull)** for PR 14731 at commit [`b63abfe`](https://github.com/apache/spark/commit/b63abfe32a5509f69c7f725a46b2e6ac8fb9cf1f).


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Is there anything else I need to do 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 pull request #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

https://github.com/apache/spark/pull/14731#discussion_r75944093

--- Diff: docs/streaming-programming-guide.md ---
@@ -644,13 +644,39 @@ methods for creating DStreams from files as input sources.
</div>
</div>

This isn't a regular expression though, what was the intent here?

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #64488 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64488/consoleFull)** for PR 14731 at commit [`fe40bd2`](https://github.com/apache/spark/commit/fe40bd2bf548ca973f9dcdf9426fb9834828f72b).
     * 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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r95058662
  
    --- Diff: docs/streaming-programming-guide.md ---
    @@ -630,35 +630,106 @@ which creates a DStream from text
     data received over a TCP socket connection. Besides sockets, the StreamingContext API provides
     methods for creating DStreams from files as input sources.
     
    -- **File Streams:** For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as:
    +#### File Streams
    +{:.no_toc}
    +
    +For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as
    +via `StreamingContext.fileStream[KeyClass, ValueClass, InputFormatClass]`.
    +
    +File streams do not require running a receiver, hence does not require allocating cores.
     
    -    <div class="codetabs">
    -    <div data-lang="scala" markdown="1">
    -        streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)
    -    </div>
    -    <div data-lang="java" markdown="1">
    -		streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory);
    -    </div>
    -    <div data-lang="python" markdown="1">
    -		streamingContext.textFileStream(dataDirectory)
    -    </div>
    -    </div>
    +For simple text files, the easiest method is `StreamingContext.textFileStream(dataDirectory)`. 
     
    -	Spark Streaming will monitor the directory `dataDirectory` and process any files created in that directory (files written in nested directories not supported). Note that
    +<div class="codetabs">
    +<div data-lang="scala" markdown="1">
     
    -     + The files must have the same data format.
    -     + The files must be created in the `dataDirectory` by atomically *moving* or *renaming* them into
    -     the data directory.
    -     + Once moved, the files must not be changed. So if the files are being continuously appended, the new data will not be read.
    +{% highlight scala %}
    +streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)
    +{% endhighlight %}
    +For text files
    +
    +{% highlight scala %}
    +streamingContext.textFileStream(dataDirectory)
    +{% endhighlight %}
    +</div>
     
    -	For simple text files, there is an easier method `streamingContext.textFileStream(dataDirectory)`. And file streams do not require running a receiver, hence does not require allocating cores.
    +<div data-lang="java" markdown="1">
    +{% highlight java %}
    +streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory);
    +{% endhighlight %}
    +For text files
     
    -	<span class="badge" style="background-color: grey">Python API</span> `fileStream` is not available in the Python API, only	`textFileStream` is	available.
    +{% highlight java %}
    +streamingContext.textFileStream(dataDirectory);
    +{% endhighlight %}
    +</div>
     
    -- **Streams based on Custom Receivers:** DStreams can be created with data streams received through custom receivers. See the [Custom Receiver
    +<div data-lang="python" markdown="1">
    +`fileStream` is not available in the Python API; only `textFileStream` is available.
    +{% highlight python %}
    +streamingContext.textFileStream(dataDirectory)
    +{% endhighlight %}
    +</div>
    +
    +</div>
    +
    +##### How Directories are Monitored
    +{:.no_toc}
    +
    +Spark Streaming will monitor the directory `dataDirectory` and process any files created in that directory.
    +
    +   * A simple directory can be monitored, such as `hdfs://namenode:8040/logs/`.
    +     All files directly under such a path will be processed as they are discovered.
    +   + A [POSIX glob pattern](http://pubs.opengroup.org/onlinepubs/009695399/utilities/xcu_chap02.html#tag_02_13_02) can be supplied, such as
    +     `hdfs://namenode:8040/logs/2017/*`.
    +     Here, the DStream will consist of all files in the directories
    +     matching the pattern.
    +     That is: it is a pattern of directories, not of files in directories.
    +   + All files must be in the same data format.
    +   * A file is considered part of a time period based on its modification time,
    +     not its creation time.
    +   + Once processed, changes to a file within the current window will not cause the file to be reread.
    +     That is: *updates are ignored*.
    +   + The more files under a directory, the longer it will take to
    +     scan for changes \u2014even if no files have been modified.
    +   * If a wildcard is used to identify directories, such as `hdfs://namenode:8040/logs/2016-*`,
    +     renaming an entire directory to match the path will add the directory to the list of
    +     monitored directories. Only the files in the directory whose modification time is
    +     within the current window will be included in the stream.
    +   + Calling `FileSystem.setTimes()` to fix the timestamp is a way to have the file picked
    +     up in a later window, even if its contents have not changed.
    +
    +
    +##### Streaming to FileSystems vs Object stores
    +{:.no_toc}
    +
    +"Full" Filesystems such as HDFS tend to set the modification time on their files as soon
    --- End diff --
    
    Is the message of this section: 
    
    Write new data to a file outside the monitored directory, and then move it into that directory after it's closed. Otherwise, if writing directly to a file in the monitored directory, it may be processed while it's being written to. However: if you are working with an object store like S3, this approach still works correctly but may be slow because a rename can be a copy. In some cases it's OK to write directly into the destination directory because the object store won't expose the file until it's done.
    
    If so I think this could be a little more direct, telling users what they need to know to use this correctly. It seems like renaming is always at least correct. I think that if different advice applies for S3, that should also be called out by name, or else I'm left wondering unnecessarily whether this applies to the most commonly-used store, S3.


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    # I'm going to scan through and tune them elsewhere; really I'm going by uses of the listFiles calls
    
    There's actually no significant use elsewhere that I can see; just a couple of uses which filter on filename \u2014so there is no cost penalty.
    
    * `SparkHadoopUtil.listLeafStatuses()` does implement its own directory recursion to find files; FileSystem.listFiles(path, true) does that, and on S3A will do flat scan that is O(files/5000); no directory overhead at all.
    * Otherwise, globStatus() can be pretty slow against object stores, but the fix there isn't in the client code; it means someone needs to implement [HADOOP-13371](https://issues.apache.org/jira/browse/HADOOP-13371), *S3A globber to use bulk listObject call over recursive directory scan* \u2014more specifically, an implementation scalable to production datasets. 
    
    Returning to this patch, should I cut out the caching? I think it is superfluous. 



---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Any more comments?


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    LGTM. Does this sort of change make sense elsewhere where `PathFilter` is used? I glanced at the others and it looked like a wash in other cases.


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r95058508
  
    --- Diff: docs/streaming-programming-guide.md ---
    @@ -630,35 +630,106 @@ which creates a DStream from text
     data received over a TCP socket connection. Besides sockets, the StreamingContext API provides
     methods for creating DStreams from files as input sources.
     
    -- **File Streams:** For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as:
    +#### File Streams
    +{:.no_toc}
    +
    +For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as
    +via `StreamingContext.fileStream[KeyClass, ValueClass, InputFormatClass]`.
    +
    +File streams do not require running a receiver, hence does not require allocating cores.
     
    -    <div class="codetabs">
    -    <div data-lang="scala" markdown="1">
    -        streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)
    -    </div>
    -    <div data-lang="java" markdown="1">
    -		streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory);
    -    </div>
    -    <div data-lang="python" markdown="1">
    -		streamingContext.textFileStream(dataDirectory)
    -    </div>
    -    </div>
    +For simple text files, the easiest method is `StreamingContext.textFileStream(dataDirectory)`. 
     
    -	Spark Streaming will monitor the directory `dataDirectory` and process any files created in that directory (files written in nested directories not supported). Note that
    +<div class="codetabs">
    +<div data-lang="scala" markdown="1">
     
    -     + The files must have the same data format.
    -     + The files must be created in the `dataDirectory` by atomically *moving* or *renaming* them into
    -     the data directory.
    -     + Once moved, the files must not be changed. So if the files are being continuously appended, the new data will not be read.
    +{% highlight scala %}
    +streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)
    +{% endhighlight %}
    +For text files
    +
    +{% highlight scala %}
    +streamingContext.textFileStream(dataDirectory)
    +{% endhighlight %}
    +</div>
     
    -	For simple text files, there is an easier method `streamingContext.textFileStream(dataDirectory)`. And file streams do not require running a receiver, hence does not require allocating cores.
    +<div data-lang="java" markdown="1">
    +{% highlight java %}
    +streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory);
    +{% endhighlight %}
    +For text files
     
    -	<span class="badge" style="background-color: grey">Python API</span> `fileStream` is not available in the Python API, only	`textFileStream` is	available.
    +{% highlight java %}
    +streamingContext.textFileStream(dataDirectory);
    +{% endhighlight %}
    +</div>
     
    -- **Streams based on Custom Receivers:** DStreams can be created with data streams received through custom receivers. See the [Custom Receiver
    +<div data-lang="python" markdown="1">
    +`fileStream` is not available in the Python API; only `textFileStream` is available.
    +{% highlight python %}
    +streamingContext.textFileStream(dataDirectory)
    +{% endhighlight %}
    +</div>
    +
    +</div>
    +
    +##### How Directories are Monitored
    +{:.no_toc}
    +
    +Spark Streaming will monitor the directory `dataDirectory` and process any files created in that directory.
    +
    +   * A simple directory can be monitored, such as `hdfs://namenode:8040/logs/`.
    +     All files directly under such a path will be processed as they are discovered.
    +   + A [POSIX glob pattern](http://pubs.opengroup.org/onlinepubs/009695399/utilities/xcu_chap02.html#tag_02_13_02) can be supplied, such as
    +     `hdfs://namenode:8040/logs/2017/*`.
    +     Here, the DStream will consist of all files in the directories
    +     matching the pattern.
    +     That is: it is a pattern of directories, not of files in directories.
    +   + All files must be in the same data format.
    +   * A file is considered part of a time period based on its modification time,
    +     not its creation time.
    +   + Once processed, changes to a file within the current window will not cause the file to be reread.
    +     That is: *updates are ignored*.
    +   + The more files under a directory, the longer it will take to
    +     scan for changes \u2014even if no files have been modified.
    +   * If a wildcard is used to identify directories, such as `hdfs://namenode:8040/logs/2016-*`,
    +     renaming an entire directory to match the path will add the directory to the list of
    +     monitored directories. Only the files in the directory whose modification time is
    +     within the current window will be included in the stream.
    +   + Calling `FileSystem.setTimes()` to fix the timestamp is a way to have the file picked
    +     up in a later window, even if its contents have not changed.
    +
    +
    +##### Streaming to FileSystems vs Object stores
    +{:.no_toc}
    +
    +"Full" Filesystems such as HDFS tend to set the modification time on their files as soon
    +as the output stream is created.
    +When a file is opened, even before data has been completely written,
    +it may be included in the `DStream` &mdash;after which updates to the file within the same window
    --- End diff --
    
    just write a hyphen here and space afterwards


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #64156 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64156/consoleFull)** for PR 14731 at commit [`b08e3c9`](https://github.com/apache/spark/commit/b08e3c9937a63a08b274a1491ea7064168646f1d).


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #64296 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64296/consoleFull)** for PR 14731 at commit [`79b57a2`](https://github.com/apache/spark/commit/79b57a2683dece86e1acd063b2d33fa5a6dd6038).


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #64296 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64296/consoleFull)** for PR 14731 at commit [`79b57a2`](https://github.com/apache/spark/commit/79b57a2683dece86e1acd063b2d33fa5a6dd6038).
     * 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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #73434 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/73434/testReport)** for PR 14731 at commit [`724495b`](https://github.com/apache/spark/commit/724495b97c1521ae5bd4c284d911c5ae6f51b19c).
     * 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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Test PASSed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/70819/
    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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r97367042
  
    --- Diff: streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala ---
    @@ -196,29 +191,29 @@ class FileInputDStream[K, V, F <: NewInputFormat[K, V]](
           logDebug(s"Getting new files for time $currentTime, " +
             s"ignoring files older than $modTimeIgnoreThreshold")
     
    -      val newFileFilter = new PathFilter {
    -        def accept(path: Path): Boolean = isNewFile(path, currentTime, modTimeIgnoreThreshold)
    -      }
    -      val directoryFilter = new PathFilter {
    -        override def accept(path: Path): Boolean = fs.getFileStatus(path).isDirectory
    -      }
    -      val directories = fs.globStatus(directoryPath, directoryFilter).map(_.getPath)
    +      val directories = Option(fs.globStatus(directoryPath)).getOrElse(Array.empty[FileStatus])
    --- End diff --
    
    I believe the getOrElse clause should handle the globStatus() return value being null, upconverting it to an empty array. Which is essentially working backwards from this special (awful) case in Glob code
    ```java
        if ((!sawWildcard) && results.isEmpty() &&
            (flattenedPatterns.size() <= 1)) {
          return null;
        }
    ```
    With the directories now being an empty array, `flatMap()` should effectively start with a `map` of the 0-element array, which is will result in the`newFiles` value also being empty. Isn't that 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 pull request #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

https://github.com/apache/spark/pull/14731#discussion_r77313878

--- Diff: docs/streaming-programming-guide.md ---
@@ -644,13 +644,44 @@ methods for creating DStreams from files as input sources.
</div>
</div>

These may be so, but what of these are actionable to a user/developer? The point about wildcards is relevant, but what can they do about slow file renames?

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #74990 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/74990/testReport)** for PR 14731 at commit [`a3aaf26`](https://github.com/apache/spark/commit/a3aaf267d2ac30c012b4a71b7a80e28a49ff10be).
     * 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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #65498 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65498/consoleFull)** for PR 14731 at commit [`735fc7c`](https://github.com/apache/spark/commit/735fc7c2343c08a323e3d213e611830e3b41ef04).
     * 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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

https://github.com/apache/spark/pull/14731#discussion_r76106355

--- Diff: docs/streaming-programming-guide.md ---
@@ -644,13 +644,39 @@ methods for creating DStreams from files as input sources.
</div>
</div>

You'll need (2015|2016) rather than (2015,2016). Also this is going to match zero or more hyphens followed by "-friday". I think you mean ".." or ".{2}" or at least ".+" instead of "*" if this is a regex.

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

https://github.com/apache/spark/pull/14731#discussion_r94407382

--- Diff: docs/streaming-programming-guide.md ---
@@ -644,17 +644,90 @@ methods for creating DStreams from files as input sources.
</div>
</div>

For simple text files, there is an easier method `streamingContext.textFileStream(dataDirectory)`. And file streams do not require running a receiver, hence does not require allocating cores.

<span class="badge" style="background-color: grey">Python API</span> `fileStream` is not available in the Python API, only `textFileStream` is available.

+
+ Special points for HDFS
--- End diff --

done

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Test PASSed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/74990/
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by uncleGen <gi...@git.apache.org>.

Github user uncleGen commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    @srowen Waiting for your final OK


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #73433 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/73433/testReport)** for PR 14731 at commit [`04f4967`](https://github.com/apache/spark/commit/04f49679b3f4f3e2d99e7cafeb9e4fa91fe98ece).
     * 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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r107140320
  
    --- Diff: streaming/src/test/scala/org/apache/spark/streaming/TestSuiteBase.scala ---
    @@ -557,4 +557,16 @@ trait TestSuiteBase extends SparkFunSuite with BeforeAndAfter with Logging {
           verifyOutput[W](output.toSeq, expectedOutput, useSet)
         }
       }
    +
    +  /**
    +   * Creates a temporary directory, which is then passed to `f` and will be deleted after `f`
    +   * returns.
    +   * (originally from `SqlTestUtils`.)
    +   * @todo Probably this method should be moved to a more general place
    +   */
    +  protected def withTempDir(f: File => Unit): Unit = {
    --- End diff --
    
    We don't already have this defined and available elsewhere?


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #64488 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64488/consoleFull)** for PR 14731 at commit [`fe40bd2`](https://github.com/apache/spark/commit/fe40bd2bf548ca973f9dcdf9426fb9834828f72b).


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r107152771
  
    --- Diff: streaming/src/test/scala/org/apache/spark/streaming/InputStreamsSuite.scala ---
    @@ -27,7 +27,8 @@ import scala.collection.JavaConverters._
     import scala.collection.mutable
     
     import com.google.common.io.Files
    -import org.apache.hadoop.fs.Path
    +import org.apache.commons.io.IOUtils
    --- End diff --
    
    I'll look @ that; I think I went with IOU as it was on the CP and I'd never had bad experiences of it. Guava, well...


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Test PASSed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64156/
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #64156 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64156/consoleFull)** for PR 14731 at commit [`b08e3c9`](https://github.com/apache/spark/commit/b08e3c9937a63a08b274a1491ea7064168646f1d).
     * 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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #73434 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/73434/testReport)** for PR 14731 at commit [`724495b`](https://github.com/apache/spark/commit/724495b97c1521ae5bd4c284d911c5ae6f51b19c).


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    @uncleGen I've updated it. Note that [HADOOP-13946](https://issues.apache.org/jira/browse/HADOOP-13946) tracks the changes in the Hadoop docs, which writes down what HDFS actually does, then note how cloud object stores have no consistent behaviour w.r.t. timestamps.While I personally believe that direct PUT calls is the way to write data, there's still ambiguity then as to when the objects get a timestamp (S3 : when the PUT/multipart put is first initiated, and not updated on the close() if the put was started earlier) \u2014so when they become visible. So: I don't go into the details, just say "look at the docs, then test on your system". That's about as authoritative as you can get


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

https://github.com/apache/spark/pull/14731#discussion_r75945790

--- Diff: docs/streaming-programming-guide.md ---
@@ -644,13 +644,39 @@ methods for creating DStreams from files as input sources.
</div>
</div>

wasn't it? My mistake. I wanted to show something fairly complex.

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

https://github.com/apache/spark/pull/14731#discussion_r77313771

--- Diff: docs/streaming-programming-guide.md ---
@@ -644,13 +644,44 @@ methods for creating DStreams from files as input sources.
</div>
</div>

It's a Hadoop-style glob rather than regular expression right? Just checking that "*" in glob would mean "1 or more characters"?

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    @uncleGen: reviewed this, tweaked the docs slightly but otherwise, there's nothing left to do that I can see


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r107140800
  
    --- Diff: streaming/src/test/scala/org/apache/spark/streaming/InputStreamsSuite.scala ---
    @@ -27,7 +27,8 @@ import scala.collection.JavaConverters._
     import scala.collection.mutable
     
     import com.google.common.io.Files
    -import org.apache.hadoop.fs.Path
    +import org.apache.commons.io.IOUtils
    --- End diff --
    
    Just use Files.write?


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Why is the caching superfluous -- because no file is evaluated more than once 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 pull request #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

https://github.com/apache/spark/pull/14731#discussion_r78932943

--- Diff: docs/streaming-programming-guide.md ---
@@ -644,13 +644,44 @@ methods for creating DStreams from files as input sources.
</div>
</div>

yes, the input is always a glob. You'd need to \ escape any file with ? in it.

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

https://github.com/apache/spark/pull/14731#discussion_r77314137

--- Diff: docs/streaming-programming-guide.md ---
@@ -644,13 +644,44 @@ methods for creating DStreams from files as input sources.
</div>
</div>

Also if I read the code correctly, it always treats the input as a glob. If so I think that's how it should be documented. That is, if you give it a path to a file containing "?" it's not going to do what you think. The glob isn't optional if I read it right.

But yeah this looks good to me code-wise.

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

https://github.com/apache/spark/pull/14731#discussion_r76105141

--- Diff: docs/streaming-programming-guide.md ---
@@ -644,13 +644,39 @@ methods for creating DStreams from files as input sources.
</div>
</div>

going with `s3a://bucket/logs/(2015,2016)-*-friday`

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Test FAILed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64534/
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #66656 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/66656/consoleFull)** for PR 14731 at commit [`f8ed8a3`](https://github.com/apache/spark/commit/f8ed8a3551d1eed5db5a22f5eeb484614036fefe).
     * 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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r103183646
  
    --- Diff: streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala ---
    @@ -140,7 +137,7 @@ class FileInputDStream[K, V, F <: NewInputFormat[K, V]](
        * a union RDD out of them. Note that this maintains the list of files that were processed
        * in the latest modification time in the previous call to this method. This is because the
        * modification time returned by the FileStatus API seems to return times only at the
    -   * granularity of seconds. And new files may have the same modification time as the
    +   * granularity of seconds in HDFS. And new files may have the same modification time as the
        * latest modification time in the previous call to this method yet was not reported in
    --- End diff --
    
    no, it really is HDFS alone. We have no idea whatsoever about the granuarity of other filesystems. Could be 2 seconds (is FAT 32 supported? Hope not) NTFS is in nanoseconds [apparently](https://msdn.microsoft.com/en-us/library/windows/desktop/ms724290(v=vs.85).aspx).


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by petermaxlee <gi...@git.apache.org>.

Github user petermaxlee commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r75583457
  
    --- Diff: streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala ---
    @@ -293,8 +290,8 @@ class FileInputDStream[K, V, F <: NewInputFormat[K, V]](
       }
     
       /** Get file mod time from cache or fetch it from the file system */
    -  private def getFileModTime(path: Path) = {
    -    fileToModTime.getOrElseUpdate(path.toString, fs.getFileStatus(path).getModificationTime())
    +  private def getFileModTime(fs: FileStatus) = {
    --- End diff --
    
    should we just remove this function 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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #64140 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64140/consoleFull)** for PR 14731 at commit [`738c51b`](https://github.com/apache/spark/commit/738c51bb57f331c58a877aa20aa5e2beb1084114).


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

https://github.com/apache/spark/pull/14731#discussion_r83491136

--- Diff: docs/streaming-programming-guide.md ---
@@ -644,17 +644,90 @@ methods for creating DStreams from files as input sources.
</div>
</div>

For simple text files, there is an easier method `streamingContext.textFileStream(dataDirectory)`. And file streams do not require running a receiver, hence does not require allocating cores.

<span class="badge" style="background-color: grey">Python API</span> `fileStream` is not available in the Python API, only `textFileStream` is available.

This is a lot of prose about HDFS semantics. Some of this seems like it's not specific to Spark's usage, and at best should be pointed to from these docs rather than repeated. What about these semantics are necessary to use this API?

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark issue #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    @srowen have you got any comments on the last patch?


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by petermaxlee <gi...@git.apache.org>.

Github user petermaxlee commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r75583436
  
    --- Diff: streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala ---
    @@ -241,16 +233,21 @@ class FileInputDStream[K, V, F <: NewInputFormat[K, V]](
        *   The files with mod time T+5 are not remembered and cannot be ignored (since, t+5 > t+1).
        *   Hence they can get selected as new files again. To prevent this, files whose mod time is more
        *   than current batch time are not considered.
    +   * @param fs file status
    +   * @param currentTime time of the batch
    +   * @param modTimeIgnoreThreshold the ignore threshold
    +   * @return true if the file has been modified within the batch window
        */
    -  private def isNewFile(path: Path, currentTime: Long, modTimeIgnoreThreshold: Long): Boolean = {
    + private def isNewFile(fs: FileStatus, currentTime: Long, modTimeIgnoreThreshold: Long): Boolean = {
    --- End diff --
    
    indent is wrong 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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Test PASSed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64296/
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #70819 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/70819/testReport)** for PR 14731 at commit [`a9a6f7b`](https://github.com/apache/spark/commit/a9a6f7b9e3876e551a2568b6220559992db40228).


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    let me do a quick review & update


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Actually, I've just noticed that DStream behaviour isn't in sync with the streaming programming guide, which says "files written in nested directories not supported)". That is: SPARK-14796 didn't patch the docs.
    
    it may as well be fixed in this patch. How about, in the bullet points underneath
    
    - Wildcards may be used to specify a set of directories to scan for new files, for example `hdfs://nn1:8050/users/alice/logs/2016-*/*.gz`
    -
    -New directories and their contents will be discovered as they arrive
    
    Special points for object stores
    - Wildcard lookup may be very slow with some object stores.
     - Directory rename is not atomic; if a directory is renamed into the streaming source, then the files within may only be discovered and process across a multiple streaming windows.
    - 
    
    + there's another optimisation; use the {{SparkHadoopUtils.isGlobPath()}} predicate to recognise when the dir path isn't a wildcard, in which case just do a simple listFiles()}}. Until that shortcutting is done automatically in the Hadoop FS implementation, spark can do it on its side. As the {{listFiles()}} call was what was used before SPARK-14796, it has to be compatible, else SPARK-14796 has broken things



---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Sean, I think I've managed to delete the lines where you were asking about globs
    
    > Am I right that the net change here is not an optimization but an expansion of the behavior to support globs rather than single dirs?
    
    There's no changes in this source to change the expansion policy; that went in with [SPARK-14976](https://issues.apache.org/jira/browse/SPARK-14976), "make StreamingContext.textFileStream support wildcard". This updates the docs to cover what goes on (the wildcard covering directories, but not the files inside them), and makes the scan much more efficient on object stores. No changes in the semantics of what gets found or when things get found


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r95058852
  
    --- Diff: streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala ---
    @@ -196,29 +191,29 @@ class FileInputDStream[K, V, F <: NewInputFormat[K, V]](
           logDebug(s"Getting new files for time $currentTime, " +
             s"ignoring files older than $modTimeIgnoreThreshold")
     
    -      val newFileFilter = new PathFilter {
    -        def accept(path: Path): Boolean = isNewFile(path, currentTime, modTimeIgnoreThreshold)
    -      }
    -      val directoryFilter = new PathFilter {
    -        override def accept(path: Path): Boolean = fs.getFileStatus(path).isDirectory
    -      }
    -      val directories = fs.globStatus(directoryPath, directoryFilter).map(_.getPath)
    +      val directories = Option(fs.globStatus(directoryPath)).getOrElse(Array.empty[FileStatus])
    --- End diff --
    
    Do you need to check for `directories.isEmpty` after this to catch this error case earlier?


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Test FAILed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64368/
    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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r75945926
  
    --- Diff: streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala ---
    @@ -196,29 +192,33 @@ class FileInputDStream[K, V, F <: NewInputFormat[K, V]](
           logDebug(s"Getting new files for time $currentTime, " +
             s"ignoring files older than $modTimeIgnoreThreshold")
     
    -      val newFileFilter = new PathFilter {
    -        def accept(path: Path): Boolean = isNewFile(path, currentTime, modTimeIgnoreThreshold)
    -      }
    -      val directoryFilter = new PathFilter {
    -        override def accept(path: Path): Boolean = fs.getFileStatus(path).isDirectory
    +      val directories = if (SparkHadoopUtil.get.isGlobPath(directoryPath)) {
    +        fs.globStatus(directoryPath)
    +            .filter(_.isDirectory)
    +            .map(_.getPath)
    +      } else {
    +        List(directoryPath).toArray
    --- End diff --
    
    yes, but obsolete right now with the refactoring in progress


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    1. updated the code to bypass the glob routine when there is no wildcard; this bypasses something fairly inefficient. 
    1. reporting FNFE on that base dir differently; skip the stack trace (maybe: log at a lower level?). 
    1. Updated the docs with a special list of blobstore best practises.
    
    It's a bit hard to get some of that phrasing of what the wildcard does right; needs careful review.
    
    Tested using my s3 streaming test, which did use a * in the wildcard. All works, but no improvements in speed on what is a fairly unrealistic structure. The time to recursively list object stores remotely is tangibly slow. Maybe that should go in the text too: "it can be take seconds to scan object stores for new data, with the time being proportional to directory depth and the number of files in a directory. Shallow and wide directory trees are faster"


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #65592 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/65592/consoleFull)** for PR 14731 at commit [`57f697d`](https://github.com/apache/spark/commit/57f697dc718e536f512c856b8e6c8239e1133fd5).


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #64486 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64486/consoleFull)** for PR 14731 at commit [`9bc0ea9`](https://github.com/apache/spark/commit/9bc0ea9734ccaf11c6306a3496c98be8cc20faab).
     * This patch **fails Scala style 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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r103184528
  
    --- Diff: docs/streaming-programming-guide.md ---
    @@ -615,35 +615,114 @@ which creates a DStream from text
     data received over a TCP socket connection. Besides sockets, the StreamingContext API provides
     methods for creating DStreams from files as input sources.
     
    -- **File Streams:** For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as:
    +#### File Streams
    +{:.no_toc}
    +
    +For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as
    +via `StreamingContext.fileStream[KeyClass, ValueClass, InputFormatClass]`.
    +
    +File streams do not require running a receiver, hence does not require allocating cores.
    +
    +For simple text files, the easiest method is `StreamingContext.textFileStream(dataDirectory)`. 
    +
    +<div class="codetabs">
    +<div data-lang="scala" markdown="1">
    +
    +{% highlight scala %}
    +streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)
    +{% endhighlight %}
    +For text files
    +
    +{% highlight scala %}
    +streamingContext.textFileStream(dataDirectory)
    +{% endhighlight %}
    +</div>
    +
    +<div data-lang="java" markdown="1">
    +{% highlight java %}
    +streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory);
    +{% endhighlight %}
    +For text files
    +
    +{% highlight java %}
    +streamingContext.textFileStream(dataDirectory);
    +{% endhighlight %}
    +</div>
     
    -    <div class="codetabs">
    -    <div data-lang="scala" markdown="1">
    -        streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)
    -    </div>
    -    <div data-lang="java" markdown="1">
    -		streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory);
    -    </div>
    -    <div data-lang="python" markdown="1">
    -		streamingContext.textFileStream(dataDirectory)
    -    </div>
    -    </div>
    +<div data-lang="python" markdown="1">
    +`fileStream` is not available in the Python API; only `textFileStream` is available.
    +{% highlight python %}
    +streamingContext.textFileStream(dataDirectory)
    +{% endhighlight %}
    +</div>
     
    -	Spark Streaming will monitor the directory `dataDirectory` and process any files created in that directory (files written in nested directories not supported). Note that
    +</div>
     
    -     + The files must have the same data format.
    -     + The files must be created in the `dataDirectory` by atomically *moving* or *renaming* them into
    -     the data directory.
    -     + Once moved, the files must not be changed. So if the files are being continuously appended, the new data will not be read.
    +##### How Directories are Monitored
    +{:.no_toc}
     
    -	For simple text files, there is an easier method `streamingContext.textFileStream(dataDirectory)`. And file streams do not require running a receiver, hence does not require allocating cores.
    +Spark Streaming will monitor the directory `dataDirectory` and process any files created in that directory.
    +
    +   * A simple directory can be monitored, such as `"hdfs://namenode:8040/logs/"`.
    +     All files directly under such a path will be processed as they are discovered.
    +   + A [POSIX glob pattern](http://pubs.opengroup.org/onlinepubs/009695399/utilities/xcu_chap02.html#tag_02_13_02) can be supplied, such as
    +     `"hdfs://namenode:8040/logs/2017/*"`.
    +     Here, the DStream will consist of all files in the directories
    +     matching the pattern.
    +     That is: it is a pattern of directories, not of files in directories.
    +   + All files must be in the same data format.
    +   * A file is considered part of a time period based on its modification time,
    +     not its creation time.
    +   + Once processed, changes to a file within the current window will not cause the file to be reread.
    +     That is: *updates are ignored*.
    +   + The more files under a directory, the longer it will take to
    +     scan for changes \u2014 even if no files have been modified.
    +   * If a wildcard is used to identify directories, such as `"hdfs://namenode:8040/logs/2016-*"`,
    +     renaming an entire directory to match the path will add the directory to the list of
    +     monitored directories. Only the files in the directory whose modification time is
    +     within the current window will be included in the stream.
    +   + Calling [`FileSystem.setTimes()`](https://hadoop.apache.org/docs/current/api/org/apache/hadoop/fs/FileSystem.html#setTimes-org.apache.hadoop.fs.Path-long-long-)
    +     to fix the timestamp is a way to have the file picked up in a later window, even if its contents have not changed.
    +
    +
    --- End diff --
    
    for "real" filesystems, rename doesn't change modtime, and files become visible in create(), so if you do a create() in the dest dir the file may be found and scanned before the data is fully written. Hence best practise: write elsewhere and rename in. Which by its very nature is the wrong thing to do for object stores: not only is there the perf hit of the copy, the timestamp changes. -and you can't set the times; if you could, we'd probably do it in the filesystem code. PUT-in-place eliminates the copy, and as the file's aren't visible until close(), no risk of early scan
    
    Of course, real (tm) filesystems do let you update the modtime, you can update them after the rename. Even better: do it before so there is no time period when they are in the dest folder with a different timestamp
    
    If you want to make sure this section is clear about what the best practises and surprises are for both FS and blobstore submission, suggest any changes


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by steveloughran <gi...@git.apache.org>.

Github user steveloughran commented on a diff in the pull request:

https://github.com/apache/spark/pull/14731#discussion_r78932779

--- Diff: docs/streaming-programming-guide.md ---
@@ -644,13 +644,44 @@ methods for creating DStreams from files as input sources.
</div>
</div>

w.r.t slow renames, I think people just need to be aware that it can be slow and windows need to plan for it. I'm soon to work on faster renames on S3 [HADOOP-13600](https://issues.apache.org/jira/browse/HADOOP-13600), but it will still be a non-atomic, O(largest-blob) operation

---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org

[GitHub] spark pull request #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by petermaxlee <gi...@git.apache.org>.

Github user petermaxlee commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r75583446
  
    --- Diff: streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala ---
    @@ -241,16 +233,21 @@ class FileInputDStream[K, V, F <: NewInputFormat[K, V]](
        *   The files with mod time T+5 are not remembered and cannot be ignored (since, t+5 > t+1).
        *   Hence they can get selected as new files again. To prevent this, files whose mod time is more
        *   than current batch time are not considered.
    +   * @param fs file status
    +   * @param currentTime time of the batch
    +   * @param modTimeIgnoreThreshold the ignore threshold
    +   * @return true if the file has been modified within the batch window
        */
    -  private def isNewFile(path: Path, currentTime: Long, modTimeIgnoreThreshold: Long): Boolean = {
    + private def isNewFile(fs: FileStatus, currentTime: Long, modTimeIgnoreThreshold: Long): Boolean = {
    --- End diff --
    
    also fs is pretty confusing, because in this context it is often used to refer to as FileSystem. We should pick a different word.



---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Test PASSed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/66656/
    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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r95058534
  
    --- Diff: docs/streaming-programming-guide.md ---
    @@ -630,35 +630,106 @@ which creates a DStream from text
     data received over a TCP socket connection. Besides sockets, the StreamingContext API provides
     methods for creating DStreams from files as input sources.
     
    -- **File Streams:** For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as:
    +#### File Streams
    +{:.no_toc}
    +
    +For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as
    +via `StreamingContext.fileStream[KeyClass, ValueClass, InputFormatClass]`.
    +
    +File streams do not require running a receiver, hence does not require allocating cores.
     
    -    <div class="codetabs">
    -    <div data-lang="scala" markdown="1">
    -        streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)
    -    </div>
    -    <div data-lang="java" markdown="1">
    -		streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory);
    -    </div>
    -    <div data-lang="python" markdown="1">
    -		streamingContext.textFileStream(dataDirectory)
    -    </div>
    -    </div>
    +For simple text files, the easiest method is `StreamingContext.textFileStream(dataDirectory)`. 
     
    -	Spark Streaming will monitor the directory `dataDirectory` and process any files created in that directory (files written in nested directories not supported). Note that
    +<div class="codetabs">
    +<div data-lang="scala" markdown="1">
     
    -     + The files must have the same data format.
    -     + The files must be created in the `dataDirectory` by atomically *moving* or *renaming* them into
    -     the data directory.
    -     + Once moved, the files must not be changed. So if the files are being continuously appended, the new data will not be read.
    +{% highlight scala %}
    +streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)
    +{% endhighlight %}
    +For text files
    +
    +{% highlight scala %}
    +streamingContext.textFileStream(dataDirectory)
    +{% endhighlight %}
    +</div>
     
    -	For simple text files, there is an easier method `streamingContext.textFileStream(dataDirectory)`. And file streams do not require running a receiver, hence does not require allocating cores.
    +<div data-lang="java" markdown="1">
    +{% highlight java %}
    +streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory);
    +{% endhighlight %}
    +For text files
     
    -	<span class="badge" style="background-color: grey">Python API</span> `fileStream` is not available in the Python API, only	`textFileStream` is	available.
    +{% highlight java %}
    +streamingContext.textFileStream(dataDirectory);
    +{% endhighlight %}
    +</div>
     
    -- **Streams based on Custom Receivers:** DStreams can be created with data streams received through custom receivers. See the [Custom Receiver
    +<div data-lang="python" markdown="1">
    +`fileStream` is not available in the Python API; only `textFileStream` is available.
    +{% highlight python %}
    +streamingContext.textFileStream(dataDirectory)
    +{% endhighlight %}
    +</div>
    +
    +</div>
    +
    +##### How Directories are Monitored
    +{:.no_toc}
    +
    +Spark Streaming will monitor the directory `dataDirectory` and process any files created in that directory.
    +
    +   * A simple directory can be monitored, such as `hdfs://namenode:8040/logs/`.
    +     All files directly under such a path will be processed as they are discovered.
    +   + A [POSIX glob pattern](http://pubs.opengroup.org/onlinepubs/009695399/utilities/xcu_chap02.html#tag_02_13_02) can be supplied, such as
    +     `hdfs://namenode:8040/logs/2017/*`.
    +     Here, the DStream will consist of all files in the directories
    +     matching the pattern.
    +     That is: it is a pattern of directories, not of files in directories.
    +   + All files must be in the same data format.
    +   * A file is considered part of a time period based on its modification time,
    +     not its creation time.
    +   + Once processed, changes to a file within the current window will not cause the file to be reread.
    +     That is: *updates are ignored*.
    +   + The more files under a directory, the longer it will take to
    +     scan for changes \u2014even if no files have been modified.
    +   * If a wildcard is used to identify directories, such as `hdfs://namenode:8040/logs/2016-*`,
    +     renaming an entire directory to match the path will add the directory to the list of
    +     monitored directories. Only the files in the directory whose modification time is
    +     within the current window will be included in the stream.
    +   + Calling `FileSystem.setTimes()` to fix the timestamp is a way to have the file picked
    --- End diff --
    
    It could be useful to link to the Hadoop javadoc for this method or at least make clear that this is an HDFS API call, not Spark


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #64140 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64140/consoleFull)** for PR 14731 at commit [`738c51b`](https://github.com/apache/spark/commit/738c51bb57f331c58a877aa20aa5e2beb1084114).
     * 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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #64662 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64662/consoleFull)** for PR 14731 at commit [`b60f175`](https://github.com/apache/spark/commit/b60f175b5ef058ed24b3ddaf9a85b899a5e33187).


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #64368 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64368/consoleFull)** for PR 14731 at commit [`b63abfe`](https://github.com/apache/spark/commit/b63abfe32a5509f69c7f725a46b2e6ac8fb9cf1f).
     * This patch **fails Spark unit 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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by SparkQA <gi...@git.apache.org>.

Github user SparkQA commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    **[Test build #64534 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/64534/consoleFull)** for PR 14731 at commit [`4134620`](https://github.com/apache/spark/commit/4134620210e28a2e182397a9bc94ccb8c4d5ffc4).
     * This patch **fails PySpark unit 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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r95058521
  
    --- Diff: docs/streaming-programming-guide.md ---
    @@ -630,35 +630,106 @@ which creates a DStream from text
     data received over a TCP socket connection. Besides sockets, the StreamingContext API provides
     methods for creating DStreams from files as input sources.
     
    -- **File Streams:** For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as:
    +#### File Streams
    +{:.no_toc}
    +
    +For reading data from files on any file system compatible with the HDFS API (that is, HDFS, S3, NFS, etc.), a DStream can be created as
    +via `StreamingContext.fileStream[KeyClass, ValueClass, InputFormatClass]`.
    +
    +File streams do not require running a receiver, hence does not require allocating cores.
     
    -    <div class="codetabs">
    -    <div data-lang="scala" markdown="1">
    -        streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)
    -    </div>
    -    <div data-lang="java" markdown="1">
    -		streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory);
    -    </div>
    -    <div data-lang="python" markdown="1">
    -		streamingContext.textFileStream(dataDirectory)
    -    </div>
    -    </div>
    +For simple text files, the easiest method is `StreamingContext.textFileStream(dataDirectory)`. 
     
    -	Spark Streaming will monitor the directory `dataDirectory` and process any files created in that directory (files written in nested directories not supported). Note that
    +<div class="codetabs">
    +<div data-lang="scala" markdown="1">
     
    -     + The files must have the same data format.
    -     + The files must be created in the `dataDirectory` by atomically *moving* or *renaming* them into
    -     the data directory.
    -     + Once moved, the files must not be changed. So if the files are being continuously appended, the new data will not be read.
    +{% highlight scala %}
    +streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)
    +{% endhighlight %}
    +For text files
    --- End diff --
    
    Space before, and colon at the end of aline?


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new...

Posted by srowen <gi...@git.apache.org>.

Github user srowen commented on a diff in the pull request:

    https://github.com/apache/spark/pull/14731#discussion_r76594233
  
    --- Diff: core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala ---
    @@ -244,6 +244,31 @@ class SparkHadoopUtil extends Logging {
       }
     
       /**
    +   * List directories/files matching the path and return the `FileStatus` results.
    +   * If the pattern is not a regexp then a simple `getFileStatus(pattern)`
    +   * is called to get the status of that path.
    +   * If the path/pattern does not match anything in the filesystem,
    +   * an empty sequence is returned.
    +   * @param pattern pattern
    +   * @return a possibly empty array of FileStatus entries
    +   */
    +  def globToFileStatus(pattern: Path): Array[FileStatus] = {
    --- End diff --
    
    Yea, but then that's wrong if for example my path actually has a ? or ^ or ] in it. It doesn't seem essential and seems even problematic to add this behavior change to an otherwise clear fix.


---
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 #14731: [SPARK-17159] [streaming]: optimise check for new files ...

Posted by AmplabJenkins <gi...@git.apache.org>.

Github user AmplabJenkins commented on the issue:

    https://github.com/apache/spark/pull/14731
  
    Test PASSed.
    Refer to this link for build results (access rights to CI server needed): 
    https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/64488/
    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