Flink - как одновременно вычислить сумму и среднее значение?

Есть ли способ во Flink (пакетная / потоковая передача) одновременно вычислять среднее значение и сумму поля? Используя метод агрегирования, я могу вычислить сумму поля для результата groupBy, но как одновременно вычислить среднее значение? Пример кода ниже.

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
DataSet<Tuple3<String,Integer,Double>> source = 
         env.readCsvFile(PathConfig.LINEITEM_1)
         .fieldDelimiter("|")
         types(String.class, Integer.class, Double.class);

source.groupBy(0,1).aggregate(Aggregations.SUM, 2);
//average of field 2???

apache-flink flink-streaming
person Eli    schedule 17.05.2017    source источник
comment
Могли бы вы использовать map / reduce вместо агрегирования?   -  person ImbaBalboa    schedule 18.05.2017
comment
Я мог бы использовать reduceGroup для вычисления суммы и среднего вручную, но, поскольку уже существует хорошая функция агрегированной суммы, я подумал, что, возможно, есть способ автоматически рассчитать среднее значение.   -  person Eli    schedule 18.05.2017

Ответы (1)


arrow_upward
2
arrow_downward

Для простых задач, таких как синтаксический анализ, группировка и агрегирование CSV, я бы рекомендовал использовать Flink API таблиц.

Если вы предпочитаете использовать более низкоуровневые API, вы можете реализовать _ 1_ функция, которая суммирует / считает (пока в итераторе не останется больше элементов) и в конце выдает окончательное среднее значение.

person twalthr    schedule 17.05.2017
comment
Я почему-то не уверен в использовании API таблиц, потому что, похоже, у него много ограничений (например, нет соединений для потоковых таблиц). Мне просто интересно, есть ли простой метод вычисления среднего, потому что это должна быть простая агрегатная функция, верно? Я видел в старом исходном коде, что действительно была средняя агрегатная функция, но она была закомментирована в новой версии, мне интересно, почему это ... - person Eli; 18.05.2017
comment
Да, у него есть ограничения, но поскольку вы можете переключаться между DataStream и Table, это довольно полезный инструмент. См. Также: flink.apache.org/news /2017/03/29/table-sql-api-update.html Усреднение кажется простым, но для этого требуются две операции (этап агрегации и окончательная карта деления), а тип возвращаемого значения зависит от ввода. Лучше позволить пользователю реализовать то, что ему нужно. - person twalthr; 23.05.2017