Same effect as {@link Iterate#groupBy(Iterable, Function)},
but executed in parallel batches, and writing output into a SynchronizedPutFastListMultimap.
505551
public static <K, V, R extends MutableMultimap<K, V>> MutableMultimap<K, V> groupBy(
            Iterable<V> iterable,
            Function<? super V, ? extends K> function,
            R concurrentMultimap,
            int batchSize)
    {
        return FJIterate.groupBy(iterable, function, concurrentMultimap, batchSize, FJIterate.FORK_JOIN_POOL);
    }
462891
public final void streamBufferWithGroups(Iterator<IN1> iterator1, Iterator<IN2> iterator2, Collector<OUT> c) {
		SingleElementPushBackIterator<IN1> i1 = new SingleElementPushBackIterator<>(iterator1);
		SingleElementPushBackIterator<IN2> i2 = new SingleElementPushBackIterator<>(iterator2);
		try {
			int size;
			if (i1.hasNext() || i2.hasNext()) {
				while (true) {
					int sig = in.readInt();
					switch (sig) {
						case SIGNAL_BUFFER_REQUEST_G0:
							if (i1.hasNext()) {
								size = sender.sendBuffer1(i1);
								sendWriteNotification(size, i1.hasNext());
							}
							break;
						case SIGNAL_BUFFER_REQUEST_G1:
							if (i2.hasNext()) {
								size = sender.sendBuffer2(i2);
								sendWriteNotification(size, i2.hasNext());
							}
							break;
						case SIGNAL_FINISHED:
							return;
						case SIGNAL_ERROR:
							try {
								outPrinter.join();
							} catch (InterruptedException e) {
								outPrinter.interrupt();
							}
							try {
								errorPrinter.join();
							} catch (InterruptedException e) {
								errorPrinter.interrupt();
							}
							throw new RuntimeException(
								"External process for task " + function.getRuntimeContext().getTaskName() + " terminated prematurely due to an error." + msg);
						default:
							receiver.collectBuffer(c, sig);
							sendReadConfirmation();
							break;
					}
				}
			}
		} catch (SocketTimeoutException ignored) {
			throw new RuntimeException("External process for task " + function.getRuntimeContext().getTaskName() + " stopped responding." + msg);
		} catch (Exception e) {
			throw new RuntimeException("Critical failure for task " + function.getRuntimeContext().getTaskName() + ". " + msg.get(), e);
		}
	}
659704
public static <K, E> Map<K, List<E>> groupBy(final Iterable<E> iterable, final Function<E, K> func) {
        final Map<K, List<E>> retVal = newLinkedHashMap();
        for (E e : iterable) {
            final K key = func.apply(e);
            List<E> val;
            if (retVal.containsKey(key)) {
                val = retVal.get(key);
            } else {
                val = newArrayList();
            }
            val.add(e);
            retVal.put(key, val);
        }
        return retVal;
    }
118432
public static <K, V> HashBagMultimap<V, K> flip(ListMultimap<K, V> listMultimap)
    {
        final HashBagMultimap<V, K> result = new HashBagMultimap<V, K>();
        listMultimap.forEachKeyMultiValues(new Procedure2<K, Iterable<V>>()
        {
            public void value(final K key, Iterable<V> values)
            {
                Iterate.forEach(values, new Procedure<V>()
                {
                    public void value(V value)
                    {
                        result.put(value, key);
                    }
                });
            }
        });
        return result;
    }
960361
public static <T> Collector<T, ArrayList<List<T>>, Stream<List<T>>> inNumBatches(int numBatches) {
        // Keep the position separate from the possible threads to enable parallel
        // collection.
        AtomicInteger nextPos = new AtomicInteger(0);
        AtomicInteger numTotal = new AtomicInteger(0);
        return Collector.of(// instantiation
                            () -> {
                                ArrayList<List<T>> batches = new ArrayList<>(numBatches);
                                for (int i = 0; i < numBatches; ++i) {
                                    batches.add(new ArrayList<>());
                                }
                                return batches;
                            },
                            // accumulator
                            (batches, item) -> {
                                int pos = nextPos.getAndUpdate(i -> (++i) % numBatches);
                                batches.get(pos).add(item);
                                numTotal.incrementAndGet();
                            },
                            // combiner
                            (a, b) -> {
                                // Merge the two lists so the batches matches the order
                                // of the non-parallel inBatchesOf with (a1..an) + (b1..bn)
                                // as the set of items. It's not extremely efficient, but
                                // works fine as this is not optimized for parallel streams.
                                for (int i = 0; i < numBatches; ++i) {
                                    List<T> al = a.get(i);
                                    List<T> bl = b.get(i);
                                    al.addAll(bl);
                                }
                                return a;
                            },
                            // finalizer
                            batches -> {
                                if (numTotal.get() < numBatches) {
                                    return batches.subList(0, numTotal.get()).stream();
                                }
                                return batches.stream();
                            });
    }
463568
public void batch(List<? extends Row> actions, @Nullable Object[] results)
      throws IOException, InterruptedException {
    if (results == null) {
      results = new Object[actions.size()];
    }
    batchCallback(actions, results, null);
  }
569809
def GroupBy(self: Iterable, f=None):
    
    if f and is_to_destruct(f):
        f = destruct_func(f)
    return _group_by(self, f)
76317
public static <T> void stream(Callback<T> callback, Iterable<T> iterable) throws Exception
    {
        for (T item : iterable) {
            try {
                callback.call(item);
            } catch (CallbackRefusedException e) {
                return;
            }
        }
    }
7503
def multi_groupby(self, key_fn):
        
        result_dict = defaultdict(list)

        for x in self:
            for key in key_fn(x):
                result_dict[key].append(x)

        # convert result lists into same Collection type as this one
        return {
            k: self.clone_with_new_elements(elements)
            for (k, elements)
            in result_dict.items()
        }
650345
private static <T> Iterable<List<T>> partitionExponentially(List<T> values, int minBatchSize, int maxBatchSize)
    {
        return () -> new AbstractIterator<List<T>>()
        {
            private int currentSize = minBatchSize;
            private final Iterator<T> iterator = values.iterator();

            @Override
            protected List<T> computeNext()
            {
                if (!iterator.hasNext()) {
                    return endOfData();
                }

                int count = 0;
                ImmutableList.Builder<T> builder = ImmutableList.builder();
                while (iterator.hasNext() && count < currentSize) {
                    builder.add(iterator.next());
                    ++count;
                }

                currentSize = min(maxBatchSize, currentSize * 2);
                return builder.build();
            }
        };
    }
1122690
function groupBy (obj, iterate, context) {
  var groupKey
  var result = {}
  if (obj) {
    context = this || context
    if (iterate && isObject(iterate)) {
      iterate = createiterateEmpty(iterate)
    } else if (!isFunction(iterate)) {
      iterate = property(iterate)
    }
    each(obj, function (val, key) {
      groupKey = iterate ? iterate.call(context, val, key, obj) : val
      if (result[groupKey]) {
        result[groupKey].push(val)
      } else {
        result[groupKey] = [val]
      }
    })
  }
  return result
}
395734
def _map_or_starmap(function, iterable, args, kwargs, map_or_starmap):
    
    arg_newarg = (("parallel", "pm_parallel"), ("chunksize", "pm_chunksize"),
                  ("pool", "pm_pool"), ("processes", "pm_processes"),
                  ("parmap_progress", "pm_pbar"))
    kwargs = _deprecated_kwargs(kwargs, arg_newarg)
    chunksize = kwargs.pop("pm_chunksize", None)
    progress = kwargs.pop("pm_pbar", False)
    progress = progress and HAVE_TQDM
    parallel, pool, close_pool = _create_pool(kwargs)
    # Map:
    if parallel:
        func_star = _get_helper_func(map_or_starmap)
        try:
            if progress and close_pool:
                try:
                    num_tasks = len(iterable)
                    # get a chunksize (as multiprocessing does):
                    chunksize = _get_default_chunksize(chunksize,
                                                       pool, num_tasks)
                    # use map_async to get progress information
                    result = pool.map_async(func_star,
                                            izip(repeat(function),
                                                 iterable,
                                                 repeat(list(args)),
                                                 repeat(kwargs)),
                                            chunksize)
                finally:
                    pool.close()
                # Progress bar:
                try:
                    _do_pbar(result, num_tasks, chunksize)
                finally:
                    output = result.get()
            else:
                result = pool.map_async(func_star,
                                        izip(repeat(function),
                                             iterable,
                                             repeat(list(args)),
                                             repeat(kwargs)),
                                        chunksize)
                output = result.get()
        finally:
            if close_pool:
                if not progress:
                    pool.close()
                pool.join()
    else:
        output = _serial_map_or_starmap(function, iterable, args, kwargs,
                                        progress, map_or_starmap)
    return output
733826
@Benchmark
  @BenchmarkMode(Mode.SampleTime)
  @OutputTimeUnit(TimeUnit.NANOSECONDS)
  public void manyRunnables() throws Exception {
    incrRunnable.val = 0;
    for (int i = 0; i < 500; i++) {
      executor.execute(incrRunnable);
    }
    executor.execute(phaserRunnable);
    phaser.arriveAndAwaitAdvance();
    if (incrRunnable.val != 500) {
      throw new AssertionError();
    }
  }
162673
public static <T> ImmutableMultimap<T, Integer> itemToIndexMultimap(
      Iterable<T> iterable) {
    final ImmutableMultimap.Builder<T, Integer> ret = ImmutableMultimap.builder();
    int idx = 0;
    for (final T x : iterable) {
      ret.put(x, idx++);
    }
    return ret.build();
  }
842744
public void collect(Object writableComparable, Object writable) throws IOException {
    if (hadoopFlowProcess instanceof HadoopFlowProcess)
      ((HadoopFlowProcess) hadoopFlowProcess).getReporter().progress();

    writer.write(writableComparable, writable);
  }
84482
@Override
	public final void flatMap(IN value, Collector<OUT> out) throws Exception {
		for (Iterator<OUT> iter = flatMap(value); iter.hasNext(); ) {
			out.collect(iter.next());
		}
	}
407559
function _groupBy(collection, getter) {
        var result = {};
        var prop;

        forEach( collection, function( elm ) {
          prop = getter(elm);

          if(!result[prop]) {
            result[prop] = [];
          }
          result[prop].push(elm);
        });
        return result;
      }