1. Utilizando a função ReduceFunction
A função ReduceFunction combina dois elementos para produzir um elemento do mesmo tipo. Ela é executada de forma incremental, após a função KeyBy.
package flink.java.test;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class ExemploReduceFunctionJanela {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment ambiente = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<tuple3>> entrada = ambiente.fromElements(DADOS_INICIAIS);
entrada.keyBy(x -> x.f0)
.countWindow(2)
.reduce(new ReduceFunction<tuple3 integer="" string="">>() {
@Override
public Tuple3<string integer="" string=""> reduce(Tuple3<string integer="" string=""> valor1, Tuple3<string integer="" string=""> valor2) throws Exception {
System.out.println("valor1-->"+valor1);
System.out.println("valor2-->"+valor2);
System.out.println("==========================");
return new Tuple3<>(valor1.f0, valor1.f1, valor1.f2 + valor2.f2);
}
}).print("reducao_acumulada");
ambiente.execute();
}
public static final Tuple3[] DADOS_INICIAIS = new Tuple3[]{
//turma nome pontuacao
Tuple3.of("turmaA","João",100),
Tuple3.of("turmaA","Maria",30),
Tuple3.of("turmaA","Pedro",70),
Tuple3.of("turmaB","Ana",50),
Tuple3.of("turmaB","Carlos",40),
Tuple3.of("turmaB","Lucas",10),
};
}
</string></string></string></tuple3></tuple3>
Resultado da execução:
valor1-->(turmaB,Ana,50)
valor2-->(turmaB,Carlos,40)
==========================
reducao_acumulada:1> (turmaB,Ana,90)
valor1-->(turmaA,João,100)
valor2-->(turmaA,Maria,30)
==========================
reducao_acumulada:2> (turmaA,João,130)
Process finished with exit code 0
2. Utilizando a função AggregateFunction para cálculos estatísticos
A função AggregateFunction é mais versátil que a ReduceFunction. Ela possui três tipos: tipo de entrada (IN), acumulador (ACC) e tipo de saída (OUT).
O tipo de entrada corersponde ao tipo do fluxo de entrada. A interface possui métodos para:
- Inicializar um acumulador
- Acumular um elemento de entrada com um acumulador existente
- Combinar dois acumuladores
- Produzir o resultado final a partir de um acumulador
entrada.keyBy(x -> x.f0)
.countWindow(2)
// AggregateFunction é mais versátil que ReduceFunction, com três tipos: entrada (IN), acumulador (ACC) e saída (OUT)
.aggregate(new AggregateFunction<tuple3 integer="" string="">, Tuple2<string>, Tuple2<string>>() {
// Criação do acumulador: inicialização do valor intermediário
@Override
public Tuple2<string integer=""> createAccumulator() {
return Tuple2.of("turmaA",1000);
}
// Operação de acumulação
@Override
public Tuple2<string integer=""> add(Tuple3<string integer="" string=""> valor1, Tuple2<string integer=""> acumulador) {
return Tuple2.of(valor1.f0, valor1.f2 + acumulador.f1);
}
// Obtenção do resultado
@Override
public Tuple2<string integer=""> getResult(Tuple2<string integer=""> acumulador) {
return Tuple2.of(acumulador.f0, acumulador.f1);
}
// Combinação de acumuladores, apenas chamada em janelas de sessão!
@Override
public Tuple2<string integer=""> merge(Tuple2<string integer=""> valor1, Tuple2<string integer=""> acumulador) {
return Tuple2.of(valor1.f0, valor1.f1 + acumulador.f1);
}
})
.print("agregacao_acumulada");
ambiente.execute();
</string></string></string></string></string></string></string></string></string></string></string></tuple3>
Resultado da execução:
agregacao_acumulada:1> (turmaB,1090)
agregacao_acumulada:2> (turmaA,1130)
Process finished with exit code 0
3. ProcessWindowFunction (Função de Janela Completa)
A função ProcessWindowFunction possui um iterável que permite acessar todos os elementos da janela. Ela também oferece um objeto de contexto para obter informações sobre tempo e estado, proporcionando maior flexibilidade em comparação com outras funções de janela.
No entanto, essa abordagem tem um custo em desempenho e recursos, pois os elementos não podem ser agregados incrementalmente. Em vez disso, ao acionar o cálculo da janela, o Flink precisa armazenar em cache todos os elementos intrenamente.
Caso 1:
entrada.keyBy(x -> x.f0)
.countWindow(2)
//public abstract class ProcessWindowFunction<in extends="" key="" out="" w="" window=""> extends AbstractRichFunction ...
.process(new ProcessWindowFunction<tuple3 integer="" string="">, Tuple3<string>, String, GlobalWindow>() {
@Override
public void process(String s, // Parâmetro 1: chave
Context contexto, // Parâmetro 2: objeto de contexto
Iterable<tuple3 integer="" string="">> elementos, // Parâmetro 3: todos os elementos desta janela
// Parâmetro 4: coletor, para传递 dados para downstream
Collector<tuple3 integer="" string="">> coletor) throws Exception {
System.out.println(contexto.window().maxTimestamp());
int soma = 0 ;
String nome = "" ;
for (Tuple3<string> tupla:elementos){
soma += tupla.f2 ;
nome = tupla.f1 ;
}
coletor.collect(Tuple3.of(s,nome,soma));
}
}).print();
</string></tuple3></tuple3></string></tuple3></in>
Rseultado da saída:
9223372036854775807
2> (turmaA,Maria,130)
9223372036854775807
1> (turmaB,Carlos,90)
Process finished with exit code 0
Caso 2:
.process(new ProcessWindowFunction<tuple2 long="">, Tuple2<string long="">, String, TimeWindow>() {
// Parâmetro 1: chave Parâmetro 2: objeto de contexto Parâmetro 3: todos os elementos nesta janela Parâmetro 4: coletor, para传递 dados para downstream
@Override
public void process(String chave,
Context contexto,
Iterable<tuple2 long="">> elementos,
Collector<tuple2 long="">> saida) throws Exception {
System.out.println(contexto.window().getStart());
long soma = 0L;
for (Tuple2<string long=""> t : elementos) {
soma += t.f1;
}
saida.collect(Tuple2.of(chave, soma));
}
})
</string></tuple2></tuple2></string></tuple2>
Referências:
- Flink Window - Funções de Redução em Janelas
- Flink(14) Funções de Janela (window function) Detalhadas