apache-flink - flink计数不同的问题
问题描述
现在我们使用翻滚窗口来计算不同的。我们遇到的问题是,如果我们将翻滚窗口从一天延长到一个月,我们就无法获得现在不同的数字。这意味着如果我们将滚动窗口设置为 1 个月,我们得到的数字是每个月的第一天。我现在如何获得当前的不同计数(现在是 3 月 9 日)?
package flink.trigger;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import java.text.SimpleDateFormat;
import java.util.Date;
public class CustomCountDistinctTigger extends Trigger<Object, TimeWindow> {
private final ReducingStateDescriptor<Long> timeState =
new ReducingStateDescriptor<>("fire-interval", new DistinctCountAggregateFunction(), LongSerializer.INSTANCE);
private long interval;
public CustomCountDistinctTigger(long interval) {
this.interval = interval;
}
@Override
public TriggerResult onElement(Object element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
ReducingState<Long> fireTimestamp = ctx.getPartitionedState(timeState);
timestamp = ctx.getCurrentProcessingTime();
if (fireTimestamp.get() == null) {
long start = timestamp - (timestamp % interval);
long nextFireTimestamp = start + interval;
ctx.registerProcessingTimeTimer(nextFireTimestamp);
fireTimestamp.add(nextFireTimestamp);
return TriggerResult.CONTINUE;
}
return TriggerResult.CONTINUE;
}
@Override
public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
// System.out.println("onProcessingTime called at "+System.currentTimeMillis() );
// return TriggerResult.FIRE_AND_PURGE;
SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
System.out.println(df.format(new Date()));
//interval
ReducingState<Long> fireTimestamp = ctx.getPartitionedState(timeState);
if(window.maxTimestamp() == time) {
return TriggerResult.FIRE_AND_PURGE;
}
else if (fireTimestamp.get().equals(time)) {
fireTimestamp.clear();
fireTimestamp.add(time + interval);
ctx.registerProcessingTimeTimer(time + interval);
return TriggerResult.FIRE;
}
return TriggerResult.CONTINUE;
}
@Override
public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
return TriggerResult.CONTINUE;
}
@Override
public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
}
}
distinct count:
DataStreamSink<Tuple2<String, Integer>> finalResultStream = keyedStream
.flatMap(new KPIDistinctDataFlatMapFunction(inputSchema))
.map(new SwapMap())
.keyBy(new WordKeySelector())
.window(TumblingProcessingTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.minutes(5)))
.trigger(new CustomCountDistinctTigger(1 * 60 * 6000))
.aggregate(new DistinctCountAggregateFunction())
.print("final print");
解决方案
您可以定义一个自定义触发器,每天返回一次 FIRE 以触发中间结果,然后在月底执行 FIRE_AND_PURGE 以关闭窗口。
每次 Trigger 返回 FIRE 时,都会通过调用 your 的process()
方法来评估您的窗口ProcessWindowFunction
,此时它可以使用Collector
提供的 生成结果。FIRE_AND_PURGE 最后一次评估窗口,然后销毁它。
另请参阅此问题的答案——如何在窗口化流 ETL 中显示中间结果?- 涵盖了一个相关主题。
推荐阅读
- reactjs - 在 gatsby jsx 中添加 `if` 条件
- php - 如何将签名添加到 XMLdocument 中的特定位置?
- typo3 - TYPO3 9.5:对单页使用内容回退模式/覆盖站点配置
- python-3.x - 将 anaconda 虚拟环境更新为特定的 python 版本
- javascript - 是否可以在命令行参数中将字符串变量作为文件传递?
- apache-flink - 如何使用 flink 在 Main Stream 中使用另一个流数据
- angularjs - 捆绑angularjs html文件.net核心
- qt - Cpp中的QThread屏障机制
- python - Scipy ODR 结果与 sd_beta 的巨大相对误差
- javascript - 如何解码和下载base64字符串(excel数据)并下载文件?