On Sat, May 27, 2023 at 4:20 PM Stephan Hoyer via dev <dev@beam.apache.org> wrote:
> On Fri, May 26, 2023 at 2:59 PM Robert Bradshaw <rober...@google.com> > wrote: > >> Yes, with_hot_key_fanout only performs a single level of fanout. I don't >> think fanning out more than this has been explored, but I would imagine >> that for most cases the increased IO would negate most if not all of the >> benefits. >> > > My reasoning for multi-level fanout would be that the total amount of IO > is that it converges as a geometric series: at each level, the amount of > data is reduced by a factor of 1/fanout. So even if fanout=2 at each level, > the total amount of IO is twice the IO of not using fanout at all. The > general IO overhead would be a factor of "fanout / (fanout - 1)". > > >> In particular, note that we already do "combiner lifting" to do as much >> combining as we can on the mapper side, e.g. suppose we have M elements and >> N workers. Each worker will (to a first order of approximation) combine M/N >> elements down to a single element, leaving N elements total to be combined >> by a worker in the subsequent stage. If N is large (or the combining >> computation expensive) one can use with_hot_key_fanout to add an >> intermediate step and let the N workers each combine M/N elements into >> sqrt(N) partial aggregates, and the subsequent worker only needs to combine >> the sqrt(N) partial aggregates. Generally N (the number of workers, not the >> number of elements) is small enough that multiple levels are not needed. >> > > Thanks for clarifying Robert. I did not realize that "combiner lifting" > was a thing! We had been operating under the assumption that we should use > fanout to sqrt(M), which could indeed be bigger than sqrt(N). In general > "with_hot_key_fanout" could use documentation to explain exactly what the > parameter means and to indicate suggested usage (e.g., set it to sqrt(N)). > > I will say that one other concern for us is memory usage. We typically > work with large, image-like data coming out of weather simulations, which > we try to divide into 10-100 MB chunks. With 1000 workers, this would > suggest fanout=30, which means combining up to 3 GB of data on a single > machine. This is probably fine but doesn't leave a large machine for error. > One thing in Robert's "to a first approximation" is that the pre-shuffle combine does flush things to shuffle to avoid running out of memory. This logic is per-SDK (because of how the combiner is invoked and also how to measure memory footprint, what caching data structure are performant, etc). So if this is working as intended, the impact of doing a large amount of pre-combining on a machine is just a lesser benefit because of having to flush more than one result per key, not excessive memory pressure. I'm sure it is inexact in specific cases, though. Kenn > > >> On Fri, May 26, 2023 at 1:57 PM Stephan Hoyer via dev < >> dev@beam.apache.org> wrote: >> >>> We have some use-cases where we are combining over very large sets >>> (e.g., computing the average of 1e5 to 1e6 elements, corresponding to >>> hourly weather observations over the past 50 years). >>> >>> "with_hot_key_fanout" seems to be rather essential for performing these >>> calculations, but as far as I can tell it only performs a single level of >>> fanout, i.e., instead of summing up 1e6 elements on a single node, you sum >>> 1e3 element 1e3 times, and then sum those 1e3 results together. >>> >>> My guess is that such combiners could be much more efficient if this was >>> performed in a hierarchical/multi-stage fashion proportional to >>> log(element_count), e.g., summing 100 elements with 3 stages, or maybe >>> summing 10 elements with 6 stages. Dask uses such a "tree reduction" >>> strategy as controlled by the "split_every" parameter: >>> https://github.com/dask/dask/blob/453bd7031828f72e4602498c1a1f776280794bea/dask/array/reductions.py#L109 >>> >>> I understand that the number of fanout stages could not be computed >>> automatically in the Beam data model, but it would still be nice to be able >>> to specify this manually. Has there been any thought to introducing this >>> feature? >>> >>> Thanks, >>> Stephan >>> >>
