Flink: DataSet.count() является узким местом - как считать параллель?

Question

Я изучаю Map-Reduce с помощью Flink и задаюсь вопросом, как эффективно подсчитывать элементы в DataSet. То, что я до сих пор это:

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

При выполнении этого, в моем Флинка войти он говорит

12/03/2016 19:47:27 DataSink (кол-()) (1/1) переключился на RUNNING

Таким образом, используется только один процессор (у меня есть четыре и другие команды, такие как уменьшение использования всех них).

Я думаю, что count() внутренне собирает DataSet со всех четырех процессоров и подсчитывает их последовательно, вместо того, чтобы каждый процессор подсчитывал свою часть, а затем суммировал ее. Это правда?

Если да, то как я могу использовать преимущества всех моих процессоров? Было бы неплохо сначала сопоставить мой DataSet с 2-кортежем, который содержит исходное значение как первый элемент, а длинное значение 1 - как второй элемент, а затем агрегировать его с помощью функции SUM?

Например, DataSet будет отображаться в DataSet>, где Long всегда будет равным 1. Поэтому, когда я суммирую все элементы, сумма второго значения кортежа будет правильным значением счетчика.

Какова наилучшая практика для подсчета элементов в DataSet?

Привет Саймон

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;