Hi all! I have flink filesource reading from an s3...
# troubleshootinga
Alex Escobedo
03/03/2023, 3:25 PMHi all! I have flink filesource reading from an s3 path. My use case is migrate the files we have from an old s3 datalake bucket in json to parquet in the new bucket.
Reading the documentation it seems like flink first has to go through all the file paths recursively to start the conversion job.
This is okay for small paths, but in prod, a few gigs was insanely long (didnt finish after going at it for over half hour) never mind a few tbs (actual goal).
Is there any way to consume the files in a reactive way (1.15)? Or is there any other technology that would benefit me better?
Alex Escobedo
03/03/2023, 3:26 PM Copy code
var source = FileSource.forRecordStreamFormat(new TextLineInputFormat(), s3Path)
.processStaticFileSet()
.setFileEnumerator((Provider) () -> new NonSplittingRecursiveEnumerator(s3FileFilter::filterPath))
.build(); Copy code
StreamingFileSink<InsightsRecord> s3Sink =
S3ParquetFileSink.createSink("output", dynamicCheckpointRollingPolicy); Copy code
env.fromSource(source, WatermarkStrategy.noWatermarks(), "s3")
.uid("read files")
.map(x -> toPojo(x))
.uid("map to json to pojo")
.addSink(s3Sink);s
Samrat Deb
03/13/2023, 8:43 AMi think you need to use batch execution mode .
https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/datastream/execution_mode/
a
Alex Escobedo
03/13/2023, 8:46 AMI did use batch but it needs to inspect the whole file system before it can start; we ended using s3 batch operations
π 1
s
Samrat Deb
03/13/2023, 8:57 AMIMO , batch jobs should take care of the splits and parallelism based on the data size.
It would be great if you can share What was the outcome of batch job ?
a
Alex Escobedo
03/13/2023, 9:04 AMYes it does take care of it, but it needs to crawl the whole file system (even with the filter) before it can start sending records to the stream. We tried with a smaller (100gb or so) s3 bucket before moving with the bigger one (few tbs)
Alex Escobedo
03/13/2023, 9:05 AMI would have liked if it processed the files as soon as they matched but from what i could see that was not the case
Alex Escobedo
03/13/2023, 9:05 AMFor more context, there were a lot of very small files
s
Samrat Deb
03/17/2023, 4:57 PMhey alex ,
i tried to repro your senerio of 1000s of small files and then do batch execution .
found out inspection takes lot of time .
i have added some debug logs and tried to calculate time taken on various steps .
this looks like a scope of improvement .
Samrat Deb
03/17/2023, 4:58 PMi will dig deeper into it more.
β
1
a
Alex Escobedo
03/17/2023, 7:34 PMI appreciate that, looking forward to follow improvements
3 Views