Good day everyone,
I have a Flink job that has an S3 folder as a source, and we keep putting
thousands of small (around 1KB each) gzip files into that folder, with the
rate of about 5000 files per minute. Here is how I created that source in
Scala:
/ val my_input_format = new TextInputFormat(new
org.apache.flink.core.fs.Path(my_path))
my_input_format.setFilesFilter(FilePathFilter.createDefaultFilter())
my_input_format.setNestedFileEnumeration(true)
val my_raw_stream = streamEnv
.readFile(my_input_format,
my_path,
FileProcessingMode.PROCESS_CONTINUOUSLY,
1000)
/
The problem is, with the monitoring interval of 1,000ms as above, about 20%
of the files were missed. From Apache Flink Dashboard, at the subsequent
operators, I could only see ~80% of the total number of files recorded
("Records sent" column).
If I increase the monitoring interval, the number of missed files would
reduce. At 5,000ms, it is about 10%, and at 30,000ms, only about 2% missed.
No WARNING/ERROR recorded though.
I could not simulate this in HDFS, as I could not reach that high file
writing speed in our cluster.
Could someone please help. Thank you very much.
--
Sent from: http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
