Flink：DataSet.count（）がボトルネックになっています - 並列数をカウントする方法は？

Question

私はFlinkを使用してMap-Reduceを学習しており、DataSetの要素を効率的に数える方法について質問があります。私はこれまで持っていることはこれです：

DataSet<MyClass> ds = ...; 
long num = ds.count(); 

これを実行すると、私のFLINKにそれが言うログ

2016年12月3日午前19時47分二十七秒DataSink（数（））（1/1）実行に切り替え

CPUが1つしか使用されていません（私は4つのコマンドを使用して、すべてのコマンドを減らしています）。

私はcount（）内部的にすべての4つのCPUからDataSetを収集し、各CPUがその部分を数えるのではなく、順番にそれらをカウントし、次にそれを合計すると思います。本当？

「はい」の場合、すべてのCPUをどのように活用できますか？データセットを、最初の項目として元の値を、2番目の項目としてlong値1を含む2タプルにマッピングしてから、SUM関数を使用して集計するとよいでしょうか？

たとえば、データセットは、Longが常に1になるDataSetにマップされます。したがって、すべてのアイテムを合計すると、タプルの2番目の値の合計が正しいカウント値になります。

データセット内のアイテムを数えるのに最適な方法は何ですか？

よろしくサイモン

Answer 1

DataSet#count()が非並列動作であるので、単一のスレッドを使用することができます。

並列化を取得するためにはカウントバイキーを行い、キーカウントには最終的な合計を適用して全体的なカウントに達し、計算時間を短縮します。

Answer 2

これは良い解決策ですか？

DataSet<Tuple1<Long>> x = ds.map(new MapFunction<MyClass, Tuple1<Long>>() { 
    @Override public Tuple1<Long> map(MyClass t) throws Exception { 
     return new Tuple1<Long>(1L); 
    } 
}).groupBy(0).sum(0); 

Long c = x.collect().iterator().next().f0;