RxはGROUP BYのCOUNTに相当しますか？

Question

私は、定期的な要素のリストを持っていると言う：

Observable<String> source = Observable.just("A", "B", "A", "C", "C", "A"); 

私は、彼らが表示された回数と一緒にその値によってグループにそれらをしたいと思いますので、出力はペアのようになります。

{"A", 3}, {"B", 1}, {"C", 2} 

基本的にSELECT x, COUNT(1) GROUP BY x;

のようなSQL文の同等は、私は彼らにGROUPBYを呼び出すためとしてのみ、これまで得た：

source.groupBy(x -> x, x -> 1) 

しかし、これはストリームをGroupedObservablesに変換するので、それらを使って移動する方法の良い例は見つかりませんでした。私はreduce()を試しましたが、groupBy()の後、GroupedObservablesを減らしたいのであって、各グループ内の要素ではないので、ここではうまくいきません。

これはGroupedObservablesで可能ですか？希望の結果を達成する他の方法も可能ですか？

Answer 1

次のコード

source.groupBy(val -> val) 
    .flatMap(
     gr -> gr.count() 
       .map(count -> new Pair<>(gr.getKey(), count) 
    ) 
).subscribe(System.out::println); 

がプリントアウトになります。

A=3 
B=1 
C=2 

Answer 2

Observable<String> source = Observable.just("A", "B", "A", "C", "C", "A"); 
    Observable<KeyValue<String, Integer>> countStream = source 
      .groupBy(val -> val) 
      .flatMap(obs -> obs.count().flatMap(cnt -> Observable.just(new KeyValue<>(obs.getKey(), cnt)))); 

    private static class KeyValue<K, V> { 

    private final K key; 
    private final V val; 

    public KeyValue(K key, V val) { 
     this.key = key; 
     this.val = val; 
    } 

    public K getKey() { 
     return key; 
    } 

    public V getVal() { 
     return val; 
    } 
} 

Answer 3

別の方法を以下に示すように、collect又はcollectInto方法を使用することです。ところで

Observable<String> source = Observable.just("A", "B", "A", "C", "C", "A"); 

source.collectInto(new HashMap<String, MutableInt>(), (map, elem) -> { 
    if (map.containsKey(elem)) { 
    map.get(elem).increment(); 
    } else { 
    map.put(elem, new MutableInt(1)); 
    } 

}).subscribe(System.out::println); 

我々はReactorを使用していた場合には多数のグループの場合には、GROUPBY後flatMapがハングアップしますので、collectは、これを行うことを示唆する方法です。 GROUPBYは、グループの低カーディナリティで最適に動作するので、それに応じてkeyMapper機能を選択したことをjavadoc of Reactor

noteから

。
特に、基準が大量のグループを生成する場合、グループが適切にダウンストリームで消費されないと（たとえば、maxConcurrencyパラメータが低すぎるflatMapによって）ハングする可能性があります。

これに関連するgithub issueもあります。