code: public <K> Stream<T> distinctBy(final Function<? super T, K> keyMapper, final Predicate<? super Long> occurrencesFilter) {
          final Supplier<? extends Map<Keyed<K, T>, Long>> supplier = isParallel() ? Suppliers.<Keyed<K, T>, Long> ofConcurrentHashMap()
                  : Suppliers.<Keyed<K, T>, Long> ofLinkedHashMap();
  
          return groupBy(Fn.<K, T> keyed(keyMapper), Collectors.counting(), supplier).filter(Fn.<Keyed<K, T>, Long> testByValue(occurrencesFilter))
                  .map(Fn.<T, K, Long> kk());
      }
docs: Distinct and filter by occurrences.
  
  @param keyMapper
  @param occurrencesFilter
  @return