Shulou(Shulou.com)06/01 Report--

This article mainly explains "how to use Flink Aggregate". Friends who are interested might as well take a look. The method introduced in this paper is simple, fast and practical. Now let the editor take you to learn how to use Flink Aggregate.

Aggregate operator: provides a function for incremental calculations based on the event window. (incrementally aggregate each data flow element in the input window and keep the window state and the elements in the window in the accumulator)

Sample environment

Java.version: 1.8.xflink.version: 1.11.1

Aggregate.java

Import com.flink.examples.DataSource;import org.apache.flink.api.common.accumulators.AverageAccumulator;import org.apache.flink.api.common.functions.AggregateFunction;import org.apache.flink.api.java.functions.KeySelector;import org.apache.flink.api.java.tuple.Tuple3;import org.apache.flink.streaming.api.datastream.DataStream;import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.util.List / * * @ Description Aggregate operator: provides a function for incremental calculation based on the event window. (incremental aggregate calculation for each data flow element in the input window, and keep the window state and elements in the window in the accumulator) * / public class Aggregate {/ * traverses the collection, printing the total number and average of different genders * @ param args * @ throws Exception * / public static void main (String [] args) throws Exception {final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment () / / Tuple3 List tuple3List = DataSource.getTuple3ToList () DataStream dataStream = env.fromCollection (tuple3List) .keyby ((KeySelector) k-> k.f1) / / scroll by quantity window, data flow for every 3 input windows Calculate once. CountWindow (3) / / can only be called based on the Windowed window Stream. Accumulator (new AggregateFunction () {/ * create a new accumulator Start aggregate calculation * @ return * / @ Override public MyAverageAccumulator createAccumulator () {return new MyAverageAccumulator () } / * add the data stream value entered by the window to the window accumulator And return the new accumulator value * @ param tuple3 * @ param accumulator * @ return * / @ Override public MyAverageAccumulator add (Tuple3 tuple3, MyAverageAccumulator accumulator) {System.out.println ("tuple3:" + tuple3.toString ()) Accumulator.setGender (tuple3.f1); / / this accumulator contains two attributes: count count and value accumulation. The add method calculates the total and summation accumulator.add (tuple3.f2) in the window; return accumulator } / * get accumulator aggregate result * @ param accumulator * @ return * / @ Override public MyAverageAccumulator getResult (MyAverageAccumulator accumulator) { Return accumulator } / * merge two accumulators Returns the status of the merged accumulator * @ param a * @ param b * @ return * / @ Override public MyAverageAccumulator merge (MyAverageAccumulator a, MyAverageAccumulator b) {a.merge (b) Return a;}}); dataStream.print (); env.execute ("flink Filter job");} / * add a gender attribute (this type is used to display the average of different genders) * / public static class MyAverageAccumulator extends AverageAccumulator {private String gender Public String getGender () {return gender;} public void setGender (String gender) {this.gender = gender;} @ Override public String toString () {/ / inherits the parent's this.getLocalValue () method to calculate and return the average return super.toString () + ", gender to" + gender;}.

Print the result

Tuple3: (Zhang San, man,20) tuple3: (Li Si, girl,24) tuple3: (Liu Liu, girl,32) tuple3: (Wang Wu, man,29) tuple3: (Wu Qi, girl,18) tuple3: (Wu Ba, man,30) 4 > AverageAccumulator 24.6666666666668 for 3 elements, gender to girl2 > AverageAccumulator 26.33333333332 for 3 elements, gender to man so far, I believe you have a deeper understanding of how Flink Aggregate is used. Here is the website, more related content can enter the relevant channels to inquire, follow us, continue to learn!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.