You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by patrickotoole <gi...@git.apache.org> on 2014/04/24 20:29:37 UTC
[GitHub] spark pull request: Add recursive directory file search to fileInp...
GitHub user patrickotoole opened a pull request:
https://github.com/apache/spark/pull/537
Add recursive directory file search to fileInputStream
Added recursive directory search to fileInputStream. Want spark to be able to find files in the subdirectories rather than just the parent directory.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/patrickotoole/spark recursive
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/537.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 #537
----
commit d2c3430155b22bec1bf8c8af3b64a38fb3c5d4c9
Author: patrickotoole <me...@rickotoole.com>
Date: 2014-04-24T18:22:16Z
Add recursive directory file search to fileinputstream
----
---
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.
---
[GitHub] spark pull request: SPARK-1795 - Add recursive directory file sear...
Posted by rajat-agarwal <gi...@git.apache.org>.
Github user rajat-agarwal commented on a diff in the pull request:
https://github.com/apache/spark/pull/537#discussion_r14364724
--- Diff: streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala ---
@@ -104,7 +122,14 @@ class FileInputDStream[K: ClassTag, V: ClassTag, F <: NewInputFormat[K,V] : Clas
logDebug("Trying to get new files for time " + currentTime)
lastNewFileFindingTime = System.currentTimeMillis
val filter = new CustomPathFilter(currentTime)
- val newFiles = fs.listStatus(directoryPath, filter).map(_.getPath.toString)
+
+ val filePaths: Array[Path] = if (recursive)
+ recursiveFileList(fs.listStatus(directoryPath).toList).toArray
--- End diff --
With below new implementation we can call this
val filePaths: Array[Path] = if (recursive)
recursiveListDirs(List(fs.getFileStatus(new Path(directoryPath)))).toArray
---
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.
---
[GitHub] spark pull request: SPARK-1795 - Add recursive directory file sear...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the pull request:
https://github.com/apache/spark/pull/537#issuecomment-96770826
Mind closing this PR?
---
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: SPARK-1795 - Add recursive directory file sear...
Posted by rajat-agarwal <gi...@git.apache.org>.
Github user rajat-agarwal commented on a diff in the pull request:
https://github.com/apache/spark/pull/537#discussion_r14324514
--- Diff: streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala ---
@@ -104,7 +122,14 @@ class FileInputDStream[K: ClassTag, V: ClassTag, F <: NewInputFormat[K,V] : Clas
logDebug("Trying to get new files for time " + currentTime)
lastNewFileFindingTime = System.currentTimeMillis
val filter = new CustomPathFilter(currentTime)
- val newFiles = fs.listStatus(directoryPath, filter).map(_.getPath.toString)
+
+ val filePaths: Array[Path] = if (recursive)
+ recursiveFileList(fs.listStatus(directoryPath).toList).toArray
--- End diff --
If the input directory is already the lowest level of directory then it will not consider any files in it.
example:
Consider the following directory.
/a/file1.txt
/a/file2.txt and so on .
If the input directory is given as "/a", there will be no output.
---
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.
---
[GitHub] spark pull request: Add recursive directory file search to fileInp...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/537#issuecomment-41315719
Can one of the admins verify this 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.
---
[GitHub] spark pull request: Add recursive directory file search to fileInp...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on the pull request:
https://github.com/apache/spark/pull/537#issuecomment-41334352
Can you please add a JIRA for this and add the JIRA number in the title, like other PRs.
---
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.
---
[GitHub] spark pull request: SPARK-1795 - Add recursive directory file sear...
Posted by rajat-agarwal <gi...@git.apache.org>.
Github user rajat-agarwal commented on a diff in the pull request:
https://github.com/apache/spark/pull/537#discussion_r14364446
--- Diff: streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala ---
@@ -97,6 +98,23 @@ class FileInputDStream[K: ClassTag, V: ClassTag, F <: NewInputFormat[K,V] : Clas
}
/**
+ * Find files recursively in a directory
+ */
+ private def recursiveFileList(
+ fileStatuses: List[FileStatus],
+ paths: List[Path] = List[Path]()
+ ): List[Path] = fileStatuses match {
+
+ case f :: tail if (fs.getContentSummary(f.getPath).getDirectoryCount > 1) =>
+ recursiveFileList(fs.listStatus(f.getPath).toList ::: tail, paths)
+ case f :: tail if f.isDir => recursiveFileList(tail, f.getPath :: paths)
+ case f :: tail => recursiveFileList(tail, paths)
+ case _ => paths
+
+ }
+
--- End diff --
how about this?
private def recursiveListDirs(fileStatuses: List[FileStatus],
paths: Set[Path] = Set[Path]()
): Set[Path] = fileStatuses match {
case f :: tail if f.isDir => recursiveListDirs(fs.listStatus(f.getPath).toList ::: tail, paths)
case f :: tail => recursiveListDirs(tail, paths + f.getPath.getParent)
case _ => paths
}
---
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.
---
[GitHub] spark pull request: Add recursive directory file search to fileInp...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on the pull request:
https://github.com/apache/spark/pull/537#issuecomment-41334446
Also, please add a unit test for this usecase in the InputStreamsSuite
---
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.
---
[GitHub] spark pull request: SPARK-1795 - Add recursive directory file sear...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/537#issuecomment-96770338
Can one of the admins verify this 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: SPARK-1795 - Add recursive directory file sear...
Posted by patrickotoole <gi...@git.apache.org>.
Github user patrickotoole commented on a diff in the pull request:
https://github.com/apache/spark/pull/537#discussion_r12507593
--- Diff: streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala ---
@@ -327,18 +327,18 @@ class StreamingContext private[streaming] (
* @param directory HDFS directory to monitor for new file
* @param filter Function to filter paths to process
* @param newFilesOnly Should process only new files and ignore existing files in the directory
+ * @param recursive Should search through the directory recursively to find new files
* @tparam K Key type for reading HDFS file
* @tparam V Value type for reading HDFS file
* @tparam F Input format for reading HDFS file
*/
def fileStream[
- K: ClassTag,
- V: ClassTag,
- F <: NewInputFormat[K, V]: ClassTag
- ] (directory: String, filter: Path => Boolean, newFilesOnly: Boolean): InputDStream[(K, V)] = {
- new FileInputDStream[K, V, F](this, directory, filter, newFilesOnly)
- }
-
+ K: ClassTag,
+ V: ClassTag,
+ F <: NewInputFormat[K, V]: ClassTag
+ ] (directory: String, filter: Path => Boolean, newFilesOnly: Boolean, recursive: Boolean): DStream[(K, V)] = {
+ new FileInputDStream[K, V, F](this, directory, filter, newFilesOnly, recursive)
+ }
--- End diff --
I have included a default value on the FileInputDStream but not on the API itself.
Wondering if we want to introduce default values to the more granular version of the API. Currently, it looks like the exposed API essentially has two versions for these method -- one that assumes default values and one that exposes all the parameters of the DStream constructor.
Thoughts?
---
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.
---
[GitHub] spark pull request: SPARK-1795 - Add recursive directory file sear...
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on the pull request:
https://github.com/apache/spark/pull/537#issuecomment-71197980
I suggest we close this in favor of https://github.com/apache/spark/pull/2765 since it implements recursion with max depth, merges, and was active more recently.
---
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: SPARK-1795 - Add recursive directory file sear...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/537
---
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: SPARK-1795 - Add recursive directory file sear...
Posted by rajat-agarwal <gi...@git.apache.org>.
Github user rajat-agarwal commented on a diff in the pull request:
https://github.com/apache/spark/pull/537#discussion_r14324543
--- Diff: streaming/src/main/scala/org/apache/spark/streaming/dstream/FileInputDStream.scala ---
@@ -97,6 +98,23 @@ class FileInputDStream[K: ClassTag, V: ClassTag, F <: NewInputFormat[K,V] : Clas
}
/**
+ * Find files recursively in a directory
+ */
+ private def recursiveFileList(
+ fileStatuses: List[FileStatus],
+ paths: List[Path] = List[Path]()
+ ): List[Path] = fileStatuses match {
+
+ case f :: tail if (fs.getContentSummary(f.getPath).getDirectoryCount > 1) =>
+ recursiveFileList(fs.listStatus(f.getPath).toList ::: tail, paths)
+ case f :: tail if f.isDir => recursiveFileList(tail, f.getPath :: paths)
+ case f :: tail => recursiveFileList(tail, paths)
+ case _ => paths
+
+ }
+
--- End diff --
If the given directory contains both files and directories then the files will be ignored and only lowest level of subdirectories will be considered.
Example:
If the input directory has following structure.
/input/directory/\<files>
/input/directory/\<directories>
And if the data that to be considered is available in files in "/input/directory/" then it will be ignored.
---
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.
---
[GitHub] spark pull request #537: SPARK-1795 - Add recursive directory file search to...
Posted by stshruthi <gi...@git.apache.org>.
Github user stshruthi commented on a diff in the pull request:
https://github.com/apache/spark/pull/537#discussion_r153595926
--- Diff: streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala ---
@@ -327,18 +327,18 @@ class StreamingContext private[streaming] (
* @param directory HDFS directory to monitor for new file
* @param filter Function to filter paths to process
* @param newFilesOnly Should process only new files and ignore existing files in the directory
+ * @param recursive Should search through the directory recursively to find new files
* @tparam K Key type for reading HDFS file
* @tparam V Value type for reading HDFS file
* @tparam F Input format for reading HDFS file
*/
def fileStream[
- K: ClassTag,
- V: ClassTag,
- F <: NewInputFormat[K, V]: ClassTag
- ] (directory: String, filter: Path => Boolean, newFilesOnly: Boolean): InputDStream[(K, V)] = {
- new FileInputDStream[K, V, F](this, directory, filter, newFilesOnly)
- }
-
+ K: ClassTag,
+ V: ClassTag,
+ F <: NewInputFormat[K, V]: ClassTag
+ ] (directory: String, filter: Path => Boolean, newFilesOnly: Boolean, recursive: Boolean): DStream[(K, V)] = {
+ new FileInputDStream[K, V, F](this, directory, filter, newFilesOnly, recursive)
+ }
--- End diff --
In which version of spark can we get the API with support for nested directory streaming?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: SPARK-1795 - Add recursive directory file sear...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/537#issuecomment-54694737
Can one of the admins verify this 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: Add recursive directory file search to fileInp...
Posted by mridulm <gi...@git.apache.org>.
Github user mridulm commented on a diff in the pull request:
https://github.com/apache/spark/pull/537#discussion_r12029172
--- Diff: streaming/src/main/scala/org/apache/spark/streaming/StreamingContext.scala ---
@@ -327,18 +327,18 @@ class StreamingContext private[streaming] (
* @param directory HDFS directory to monitor for new file
* @param filter Function to filter paths to process
* @param newFilesOnly Should process only new files and ignore existing files in the directory
+ * @param recursive Should search through the directory recursively to find new files
* @tparam K Key type for reading HDFS file
* @tparam V Value type for reading HDFS file
* @tparam F Input format for reading HDFS file
*/
def fileStream[
- K: ClassTag,
- V: ClassTag,
- F <: NewInputFormat[K, V]: ClassTag
- ] (directory: String, filter: Path => Boolean, newFilesOnly: Boolean): InputDStream[(K, V)] = {
- new FileInputDStream[K, V, F](this, directory, filter, newFilesOnly)
- }
-
+ K: ClassTag,
+ V: ClassTag,
+ F <: NewInputFormat[K, V]: ClassTag
+ ] (directory: String, filter: Path => Boolean, newFilesOnly: Boolean, recursive: Boolean): DStream[(K, V)] = {
+ new FileInputDStream[K, V, F](this, directory, filter, newFilesOnly, recursive)
+ }
--- End diff --
This looks like an api change - please add default value to recursive
---
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.
---
[GitHub] spark pull request: SPARK-1795 - Add recursive directory file sear...
Posted by tdas <gi...@git.apache.org>.
Github user tdas commented on the pull request:
https://github.com/apache/spark/pull/537#issuecomment-68079080
@patrickotoole Sorry for this patch sitting around here for so long without any attention. Mind updating this patch to the latest code.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org