Thanks Arvid! Will try to increase the property you recommended and will post the update.
On Sat, Jun 6, 2020, 7:33 AM Arvid Heise <ar...@ververica.com> wrote: > Hi Venkata, > > you can find them on the Hadoop AWS page (we are just using it as a > library) [1]. > > [1] > https://hadoop.apache.org/docs/current/hadoop-aws/tools/hadoop-aws/index.html#General_S3A_Client_configuration > > On Sat, Jun 6, 2020 at 1:26 AM venkata sateesh` kolluru < > vkollur...@gmail.com> wrote: > >> Hi Kostas and Arvid, >> >> Thanks for your suggestions. >> >> The small files were already created and I am trying to roll few into a >> big file while sinking. But due to the custom bucket assigner, it is hard >> getting more files with in the same prefix in specified checkinpointing >> time. >> >> For example: >> <bucket>/prefix1/prefix2/YY/MM/DD/HH is our structure in s3. >> checkpointing interval is 5 minutes. prefix1 has 40 different values and >> prefix 2 has 10000+ values >> With in the 5 minute interval, we are able to get part file size in these >> prefixes not more than 5-10 files. >> >> Regarding printstream, will figure out how to use SimpleStringEncoder on >> a Tuple as I only need to write tuple.f2 element in the file. If you can >> guide me on how to do it, I would appreciate it. >> >> Will try Arvid suggestion on increasing fs.s3a.connection.maximum . I >> was trying to find about these parameters and could find anywhere. Is there >> a place that I could look at these config params list ? >> >> Also I am using s3:// as prefix, would fs.s3a.connection.maximum affect >> that too or is there separate param like fs.s3.connection.maximum. >> >> On Fri, Jun 5, 2020 at 2:13 PM Kostas Kloudas <kklou...@gmail.com> wrote: >> >>> Hi all, >>> >>> @Venkata, Do you have many small files being created as Arvid suggested? >>> If yes, then I tend to agree that S3 is probably not the best sink. >>> Although I did not get that from your description. >>> In addition, instead of PrintStream you can have a look at the code of >>> the SimpleStringEncoder in Flink [1] for a bit more efficient >>> implementation. >>> >>> Cheers, >>> Kostas >>> >>> [1] >>> https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/api/common/serialization/SimpleStringEncoder.java >>> >>> >>> On Fri, Jun 5, 2020 at 7:56 PM Arvid Heise <ar...@ververica.com> wrote: >>> >>>> Hi Venkata, >>>> >>>> are the many small files intended or is it rather an issue of our >>>> commit on checkpointing? If so then FLINK-11499 [1] should help you. Design >>>> is close to done, unfortunately implementation will not make it into 1.11. >>>> >>>> In any case, I'd look at the parameter fs.s3a.connection.maximum, as >>>> you store both state and data on S3. I'd probably go with slot*3 or even >>>> higher. >>>> >>>> Lastly, the way you output elements looks also a bit suspicious. >>>> PrintStream is not known for great performance. I'm also surprised that it >>>> works without manual flushing. >>>> >>>> [1] https://issues.apache.org/jira/browse/FLINK-11499 >>>> >>>> On Mon, Jun 1, 2020 at 9:52 AM Jörn Franke <jornfra...@gmail.com> >>>> wrote: >>>> >>>>> I think S3 is a wrong storage backend for this volumes of small >>>>> messages. >>>>> Try to use a NoSQL database or write multiple messages into one file >>>>> in S3 (10000 or 100000) >>>>> >>>>> If you still want to go with your scenario then try a network >>>>> optimized instance and use s3a in Flink and configure s3 entropy. >>>>> >>>>> Am 31.05.2020 um 15:30 schrieb venkata sateesh` kolluru < >>>>> vkollur...@gmail.com>: >>>>> >>>>> >>>>> Hi David, >>>>> >>>>> The avg size of each file is around 30KB and I have checkpoint >>>>> interval of 5 minutes. Some files are even 1 kb, because of checkpoint >>>>> some >>>>> files are merged into 1 big file around 300MB. >>>>> >>>>> With 120 million files and 4Tb, if the rate of transfer is 300 per >>>>> minute, it is taking weeks to write to s3. >>>>> >>>>> I have tried to increase parallelism of sink but I dont see any >>>>> improvement. >>>>> >>>>> The sink record is Tuple3<String,String,String>, the actual content of >>>>> file is f2. This is content is written to <s3 bucket>/f0/f1/part*-* >>>>> >>>>> I guess the prefix determination in custombucketassigner wont be >>>>> causing this delay? >>>>> >>>>> Could you please shed some light on writing custom s3 sink ? >>>>> >>>>> Thanks >>>>> >>>>> >>>>> On Sun, May 31, 2020, 6:34 AM David Magalhães <speeddra...@gmail.com> >>>>> wrote: >>>>> >>>>>> Hi Venkata. >>>>>> >>>>>> 300 requests per minute look like a 200ms per request, which should >>>>>> be a normal response time to send a file if there isn't any speed >>>>>> limitation (how big are the files?). >>>>>> >>>>>> Have you changed the parallelization to be higher than 1? I also >>>>>> recommend to limit the source parallelization, because it can consume >>>>>> pretty fast from Kafka and create some kind of backpressure. >>>>>> >>>>>> I don't any much experience with StreamingFileSink, because I've >>>>>> ended up using a custom S3Sink, but I did have some issues writing to S3 >>>>>> because the request wasn't parallelised. Check this thread, >>>>>> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Parallisation-of-S3-write-sink-td34064.html#a34070 >>>>>> >>>>>> On Sun, May 31, 2020 at 1:32 AM venkata sateesh` kolluru < >>>>>> vkollur...@gmail.com> wrote: >>>>>> >>>>>>> Hello, >>>>>>> >>>>>>> I have posted the same in stackoverflow but didnt get any response. >>>>>>> So posting it here for help. >>>>>>> >>>>>>> >>>>>>> https://stackoverflow.com/questions/62068787/flink-s3-write-performance-optimization?noredirect=1#comment109814428_62068787 >>>>>>> >>>>>>> Details: >>>>>>> >>>>>>> I am working on a flink application on kubernetes(eks) which >>>>>>> consumes data from kafka and write it to s3. >>>>>>> >>>>>>> We have around 120 million xml messages of size 4TB in kafka. >>>>>>> Consuming from kafka is super fast. >>>>>>> >>>>>>> These are just string messages from kafka. >>>>>>> >>>>>>> There is a high back pressure while writing to s3. We are not even >>>>>>> hitting the s3 PUT request limit which is arounf 3500 requests/sec. I am >>>>>>> seeing only 300 writes per minute to S3 which is very slow. >>>>>>> >>>>>>> I am using StreamFileSink to write to s3 with Rolling policy as >>>>>>> OnCheckpointPolicy. >>>>>>> >>>>>>> Using flink-s3-fs-hadoop-*.jar and s3:// as path (not using any s3a >>>>>>> or s3p) >>>>>>> >>>>>>> Other than this I dont have any config related to s3 >>>>>>> >>>>>>> StreamingFileSink<Tuple3<String,String, String>> sink = >>>>>>> StreamingFileSink >>>>>>> .forRowFormat(new Path(s3://BUCKET), >>>>>>> (Tuple3<String,String, String> element, >>>>>>> OutputStream stream) -> { >>>>>>> PrintStream out = new PrintStream(stream); >>>>>>> out.println(element.f2); >>>>>>> }) >>>>>>> // Determine component type for each record >>>>>>> .withBucketAssigner(new CustomBucketAssigner()) >>>>>>> .withRollingPolicy(OnCheckpointRollingPolicy.build()) >>>>>>> .withBucketCheckInterval((TimeUnit.MINUTES.toMillis(1))) >>>>>>> .build(); >>>>>>> >>>>>>> Is there anything that we can optimize on s3 from streamfilesink or >>>>>>> in flink-conf.xml ? >>>>>>> >>>>>>> Like using bulkformat or any config params like fs.s3.maxThreads etc. >>>>>>> >>>>>>> For checkpointing too I am using s3:// instead of s3p or s3a >>>>>>> >>>>>>> env.setStateBackend((StateBackend) new >>>>>>> RocksDBStateBackend(s3://checkpoint_bucket, true)); >>>>>>> env.enableCheckpointing(300000); >>>>>>> >>>>>>> >>>>>>> >>>> >>>> -- >>>> >>>> Arvid Heise | Senior Java Developer >>>> >>>> <https://www.ververica.com/> >>>> >>>> Follow us @VervericaData >>>> >>>> -- >>>> >>>> Join Flink Forward <https://flink-forward.org/> - The Apache Flink >>>> Conference >>>> >>>> Stream Processing | Event Driven | Real Time >>>> >>>> -- >>>> >>>> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany >>>> >>>> -- >>>> Ververica GmbH >>>> Registered at Amtsgericht Charlottenburg: HRB 158244 B >>>> Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji >>>> (Toni) Cheng >>>> >>> > > -- > > Arvid Heise | Senior Java Developer > > <https://www.ververica.com/> > > Follow us @VervericaData > > -- > > Join Flink Forward <https://flink-forward.org/> - The Apache Flink > Conference > > Stream Processing | Event Driven | Real Time > > -- > > Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany > > -- > Ververica GmbH > Registered at Amtsgericht Charlottenburg: HRB 158244 B > Managing Directors: Timothy Alexander Steinert, Yip Park Tung Jason, Ji > (Toni) Cheng >
