Hi Robert, Thanks for the input. I did increase the amount of managed memory, and confirmed that both SSDs (on each slave) are being used for temp data.
I haven’t been able to figure out why the server CPU usage is low, but I did notice that it fluctuated from very low (10%) on up to 95+%, with the average around 50%. But iowait never gets very high. Wondering if CPU is low when a lot of segments are being flushed to disk, and high when a lot of segments are being sorted before being flushed. The main bottleneck is the CoGroup operation, which is in the phase where it's writing a all of the (grouped) data to disk, in preparation for the sorted merge to do the grouping. Looking at threads from a single dump of a TM process, most are WAITING, with counts like: 47 - requestMemorySegmentBlocking 70 - ReaderIterator.next 70 - AbstractRecordReader.getNextRecord The only RUNNABLE threads that were doing anything interesting were all Kryo-related, which speaks to your point about ensuring I’m using POJOs. I’m curious, after looking into the code, whether enabling object reuse would also help - I see different versions of mergers being used, depending on whether that’s on or not. Thanks again, — Ken > On Sep 11, 2020, at 5:27 AM, Robert Metzger <rmetz...@apache.org> wrote: > > Hi Ken, > > Some random ideas that pop up in my head: > - make sure you use data types that are efficient to serialize, and cheap to > compare (ideally use primitive types in TupleN or POJOs) > - Maybe try the TableAPI batch support (if you have time to experiment). > - optimize memory usage on the TaskManager for a lot of managed memory on the > TaskManager, so that we have more memory for efficient sorting (leading to > less spilling): > https://ci.apache.org/projects/flink/flink-docs-release-1.11/ops/memory/mem_tuning.html#configure-memory-for-batch-jobs > > <https://ci.apache.org/projects/flink/flink-docs-release-1.11/ops/memory/mem_tuning.html#configure-memory-for-batch-jobs> > - make sure to configure a separate tmp directory for each SSD, so that we > can spread the load across all SSDs. > - If you are saying the CPU load is 40% on a TM, we have to assume we are IO > bound: Is it the network or the disk(s)? > > I hope this is some helpful inspiration for improving the performance. > > > On Fri, Sep 4, 2020 at 9:43 PM Ken Krugler <kkrugler_li...@transpac.com > <mailto:kkrugler_li...@transpac.com>> wrote: > Hi all, > > I added a CoGroup to my batch job, and it’s now running much slower, > primarily due to back pressure from the CoGroup operator. > > I assume it’s because this operator is having to sort/buffer-to-disk all > incoming data. Looks like about 1TB from one side of the join, currently very > little from the other but will be up to 2TB in the future. > > I don’t see lots of GC, I’m using about 60% of available network buffers, per > TM server load (for all 8 servers) is about 40% average, and both SSDs on > each TM are being used for …/flink-io-xxx/yyy.channel files. > > What are techniques for improving the performance of a CoGroup? > > Thanks! > > — Ken > > -------------------------- > Ken Krugler > http://www.scaleunlimited.com <http://www.scaleunlimited.com/> > custom big data solutions & training > Hadoop, Cascading, Cassandra & Solr > -------------------------- Ken Krugler http://www.scaleunlimited.com custom big data solutions & training Hadoop, Cascading, Cassandra & Solr
