Hi Niklas,
I don't know exactly what is going wrong there, but I have a few pointers
for you:
1) in cluster setups, Flink redirects println() to ./log/*.out files, i.e,
you have to search for the task manager that ran the DirReader and check
its ./log/*.out file
2) you are using Java's File class. That will only work if you are
accessing the local file system of the machine the DirReader runs on. If
you want to read the files from an HDFS you have to use the corresponding
HDFS client.
3) I would not extend the FileInputFormat for your purpose. The
FileInputFormat is meant to *read* files, not just look up file names. I'd
rather implement an InputFormat from scratch. Since you are only running a
single instance, you can return a single dummy InputSplit.
Let me know, if you have further questions.
Best, Fabian
2016-10-28 18:38 GMT+02:00 Niklas Semmler <nik...@inet.tu-berlin.de>:
> Hello Flink community,
>
> I am running into an issue with a custom FileInputFormat class and would
> appreciate your help.
>
> My goal is to read all files from a directory as paths:
>
> val env : ExecutionEnvironment = ExecutionEnvironment.getExecut
> ionEnvironment
>
> var source : DataSet[String] = env.readFile(new DirReader,
> "/tmp/mydir").setParallelism(1)
>
> source.writeAsText("/tmp/results", WriteMode.OVERWRITE)
>
> env.execute("Job")
>
> It works, when I execute the program from within my IDE or execute it
> directly as a fat jar. When I run it through the Flink CLI the file
> "/tmp/results" is created, but not filled with entries.
>
> There seems to be something wrong with my custom DirReader (see below).
> The output of the println statements is not visible when running the code
> from the Flink CLI.
>
> No exception is stated in the logs (see below). I am at a loss at what to
> try. Even worse, when I copy the fat jar to a remote system, the problem
> appears also when I execute the fat jar directly.
>
> Local System
> Flink: 1.0.2
> Java: 1.8.0_102
> Scala: 2.11.8
>
> Remote System
> Flink: 1.1.3
> Java: 1.8.0_92
> Scala: 2.11.6
>
> Help or ideas to try out are welcome!
>
> Best,
> Niklas
>
>
>
> ----------------------------------------
> import java.io.File
>
> import org.apache.flink.api.common.io.FileInputFormat
>
> class DirReader extends FileInputFormat[String] {
> var running : Boolean = false
> var fileList : Array[String] = null
>
> override def openInputFormat() = {
> println("Path: " + this.filePath.toString)
> val directory = new File(this.filePath.toString)
> if (directory != null && directory.isDirectory) {
> fileList = directory.listFiles.filter(_.i
> sDirectory).map(_.listFiles).flatten
> .map(_.toString)
> running = if (fileList.length > 1) true else false
> }
> println("fileList " + fileList.length + " running " + running)
> }
>
> override def nextRecord(reuse: String): String = {
> val head = fileList.head
> println("File: " + head)
> fileList = fileList.tail
> running = if (fileList.length == 0) false else true
> head
> }
>
> override def reachedEnd(): Boolean = ! running
> }
> ----------------------------------------
>
> The output from the CLI:
>
> 10/28/2016 18:27:56 Job execution switched to status RUNNING.
> 10/28/2016 18:27:56 DataSource (at org.apache.flink.api.scala.Exe
> cutionEnvironment.readFile(ExecutionEnvironment.scala:385)
> (de.tuberlin.inet.plag.DirReader))(1/1) switched to SCHEDULED
> 10/28/2016 18:27:56 DataSource (at org.apache.flink.api.scala.Exe
> cutionEnvironment.readFile(ExecutionEnvironment.scala:385)
> (de.tuberlin.inet.plag.DirReader))(1/1) switched to DEPLOYING
> 10/28/2016 18:27:56 DataSource (at org.apache.flink.api.scala.Exe
> cutionEnvironment.readFile(ExecutionEnvironment.scala:385)
> (de.tuberlin.inet.plag.DirReader))(1/1) switched to RUNNING
> 10/28/2016 18:27:56 DataSink (TextOutputFormat (/tmp/results) -
> UTF-8)(1/1) switched to SCHEDULED
> 10/28/2016 18:27:56 DataSink (TextOutputFormat (/tmp/results) -
> UTF-8)(1/1) switched to DEPLOYING
> 10/28/2016 18:27:56 DataSource (at org.apache.flink.api.scala.Exe
> cutionEnvironment.readFile(ExecutionEnvironment.scala:385)
> (de.tuberlin.inet.plag.DirReader))(1/1) switched to FINISHED
> 10/28/2016 18:27:56 DataSink (TextOutputFormat (/tmp/results) -
> UTF-8)(1/1) switched to RUNNING
> 10/28/2016 18:27:56 DataSink (TextOutputFormat (/tmp/results) -
> UTF-8)(1/1) switched to FINISHED
> 10/28/2016 18:27:56 Job execution switched to status FINISHED.
>
>
> --
> Niklas Semmler
> PhD Student / Research Assistant
> TU Berlin, INET, Room MAR 4.027
> Marchstr 23, 10587 Berlin
> Tel.: +49 (0)30 314 75739
> http://inet.tu-berlin.de/~nsemmler/
>
