diff --git a/docs/programming_guide.md b/docs/programming_guide.md
index 6c32250bbf6c5ee30ff9911a3b7518dd221a73cd..5a8614aaf3497efa066faa4a11d0a42347ebc0b5 100644
--- a/docs/programming_guide.md
+++ b/docs/programming_guide.md
@@ -1584,7 +1584,7 @@ Data Sources
Data sources create the initial data sets, such as from files or from Java collections. The general
-mechanism of of creating data sets is abstracted behind an
+mechanism of creating data sets is abstracted behind an
{% gh_link /flink-core/src/main/java/org/apache/flink/api/common/io/InputFormat.java "InputFormat"%}.
Flink comes
with several built-in formats to create data sets from common file formats. Many of them have
@@ -1662,11 +1662,30 @@ DataSet dbData =
// manually provide the type information as shown in the examples above.
{% endhighlight %}
+#### Recursive Traversal of the Input Path Directory
+
+For file-based inputs, when the input path is a directory, nested files are not enumerated by default. Instead, only the files inside the base directory are read, while nested files are ignored. Recursive enumeration of nested files can be enabled through the `recursive.file.enumeration` configuration parameter, like in the following example.
+
+{% highlight java %}
+// enable recursive enumeration of nested input files
+ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+// create a configuration object
+Configuration parameters = new Configuration();
+
+// set the recursive enumeration parameter
+parameters.setBoolean("recursive.file.enumeration", true);
+
+// pass the configuration to the data source
+DataSet logs = env.readTextFile("file:///path/with.nested/files")
+ .withParameters(parameters);
+{% endhighlight %}
+
Data sources create the initial data sets, such as from files or from Java collections. The general
-mechanism of of creating data sets is abstracted behind an
+mechanism of creating data sets is abstracted behind an
{% gh_link /flink-core/src/main/java/org/apache/flink/api/common/io/InputFormat.java "InputFormat"%}.
Flink comes
with several built-in formats to create data sets from common file formats. Many of them have
@@ -1730,9 +1749,27 @@ val values = env.fromElements("Foo", "bar", "foobar", "fubar")
// generate a number sequence
val numbers = env.generateSequence(1, 10000000);
{% endhighlight %}
+
+#### Recursive Traversal of the Input Path Directory
+
+For file-based inputs, when the input path is a directory, nested files are not enumerated by default. Instead, only the files inside the base directory are read, while nested files are ignored. Recursive enumeration of nested files can be enabled through the `recursive.file.enumeration` configuration parameter, like in the following example.
+
+{% highlight scala %}
+// enable recursive enumeration of nested input files
+val env = ExecutionEnvironment.getExecutionEnvironment
+
+// create a configuration object
+val parameters = new Configuration
+
+// set the recursive enumeration parameter
+parameters.setBoolean("recursive.file.enumeration", true)
+
+// pass the configuration to the data source
+env.readTextFile("file:///path/with.nested/files").withParameters(parameters)
+{% endhighlight %}
+
-
[Back to top](#top)
Data Sinks
diff --git a/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java b/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java
index 30426f6b046c9921315d66947c4a496022cc1179..2f3f59f25ccea45a1d1314c1ba95294c4f2da8bf 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/io/FileInputFormat.java
@@ -66,7 +66,7 @@ public abstract class FileInputFormat implements InputFormat implements InputFormat implements InputFormat implements InputFormat implements InputFormat implements InputFormat implements InputFormat implements InputFormat implements InputFormat files, long length)
+ throws IOException {
+ final FileSystem fs = path.getFileSystem();
+
+ for(FileStatus dir: fs.listStatus(path)) {
+ if (dir.isDir()) {
+ addNestedFiles(dir.getPath(), files, length);
+ }
+ else {
+ if(acceptFile(dir)) {
+ files.add(dir);
+ length += dir.getLen();
+ testForUnsplittable(dir);
+ }
+ }
+ }
+ return length;
+ }
+
private boolean testForUnsplittable(FileStatus pathFile) {
if(pathFile.getPath().getName().endsWith(DEFLATE_SUFFIX)) {
unsplittable = true;
@@ -807,6 +853,11 @@ public abstract class FileInputFormat implements InputFormat {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public boolean reachedEnd() throws IOException {
+ return true;
+ }
+
+ @Override
+ public IntValue nextRecord(IntValue reuse) throws IOException {
+ return null;
+ }
+ }
+}
\ No newline at end of file
diff --git a/flink-core/src/test/java/org/apache/flink/api/common/io/FileInputFormatTest.java b/flink-core/src/test/java/org/apache/flink/api/common/io/FileInputFormatTest.java
index d6c21f40ecaed87eb60705a8069c21e71210e4fd..58154e50cd5be382c1600b6c3e4dbc501c5bbbc4 100644
--- a/flink-core/src/test/java/org/apache/flink/api/common/io/FileInputFormatTest.java
+++ b/flink-core/src/test/java/org/apache/flink/api/common/io/FileInputFormatTest.java
@@ -18,8 +18,10 @@
package org.apache.flink.api.common.io;
+import java.io.BufferedOutputStream;
import java.io.BufferedWriter;
import java.io.File;
+import java.io.FileOutputStream;
import java.io.FileWriter;
import java.io.IOException;
import java.net.URI;
@@ -288,6 +290,55 @@ public class FileInputFormatTest {
Assert.fail(e.getMessage());
}
}
+
+ @Test
+ public void testGetStatsIgnoredUnderscoreFiles() {
+ try {
+ final long SIZE = 2048;
+ final long TOTAL = 2*SIZE;
+
+ // create two accepted and two ignored files
+ File tempDir = new File(System.getProperty("java.io.tmpdir"));
+ File f = null;
+ do {
+ f = new File(tempDir, TestFileUtils.randomFileName(""));
+ } while (f.exists());
+ f.mkdirs();
+ f.deleteOnExit();
+
+ File child1 = new File(f, "dataFile1.txt");
+ File child2 = new File(f, "another_file.bin");
+ File luigiFile = new File(f, "_luigi");
+ File success = new File(f, "_SUCCESS");
+
+ File[] files = { child1, child2, luigiFile, success };
+
+ for (File child : files) {
+ child.deleteOnExit();
+
+ BufferedOutputStream out = new BufferedOutputStream(new FileOutputStream(child));
+ try {
+ for (long bytes = SIZE; bytes > 0; bytes--) {
+ out.write(0);
+ }
+ } finally {
+ out.close();
+ }
+ }
+ final DummyFileInputFormat format = new DummyFileInputFormat();
+ format.setFilePath(f.toURI().toString());
+ format.configure(new Configuration());
+
+ // check that only valid files are used for statistics computation
+ BaseStatistics stats = format.getStatistics(null);
+ Assert.assertEquals(TOTAL, stats.getTotalInputSize());
+ }
+ catch (Exception e) {
+ System.err.println(e.getMessage());
+ e.printStackTrace();
+ Assert.fail(e.getMessage());
+ }
+ }
// ------------------------------------------------------------------------
diff --git a/flink-core/src/test/java/org/apache/flink/testutils/TestFileUtils.java b/flink-core/src/test/java/org/apache/flink/testutils/TestFileUtils.java
index 83835ca46cf525934de723ebbef36ffcdb093aeb..683bc4d30e12982150954bbed4f659e53836e166 100644
--- a/flink-core/src/test/java/org/apache/flink/testutils/TestFileUtils.java
+++ b/flink-core/src/test/java/org/apache/flink/testutils/TestFileUtils.java
@@ -63,7 +63,27 @@ public class TestFileUtils {
}
return f.toURI().toString();
}
-
+
+ public static String createTempFileInDirectory(String dir, long bytes) throws IOException {
+ File f;
+ do {
+ f = new File(dir + "/" + randomFileName());
+ } while (f.exists());
+ f.getParentFile().mkdirs();
+ f.createNewFile();
+ f.deleteOnExit();
+
+ BufferedOutputStream out = new BufferedOutputStream(new FileOutputStream(f));
+ try {
+ for (; bytes > 0; bytes--) {
+ out.write(0);
+ }
+ } finally {
+ out.close();
+ }
+ return f.toURI().toString();
+ }
+
public static String createTempFile(String contents) throws IOException {
File f = File.createTempFile(FILE_PREFIX, FILE_SUFFIX);
f.deleteOnExit();