在CoFlatMapFunction
方面遇到了一些困难。如果我将它放在窗口之前的DataStream
上,它似乎可以正常工作,但如果将其放在窗口的“apply”函数之后,就会出现错误。
我正在测试两个流,主要的“Features”在flatMap1
上不断地输入数据,控制流“Model”在flatMap2
上通过请求更改模型。
我能够在flatMap2
中正确设置并看到b0/b1的值,但是flatMap1
始终将b0和b1视为初始化时设置为0的值。
我是否遗漏了一些明显的东西?
public static class applyModel implements CoFlatMapFunction<Features, Model, EnrichedFeatures> {
private static final long serialVersionUID = 1L;
Double b0;
Double b1;
public applyModel(){
b0=0.0;
b1=0.0;
}
@Override
public void flatMap1(Features value, Collector<EnrichedFeatures> out) {
System.out.print("Main: " + this + "\n");
}
@Override
public void flatMap2(Model value, Collector<EnrichedFeatures> out) {
System.out.print("Old Model: " + this + "\n");
b0 = value.getB0();
b1 = value.getB1();
System.out.print("New Model: " + this + "\n");
}
@Override
public String toString(){
return "CoFlatMapFunction: {b0: " + b0 + ", b1: " + b1 + "}";
}
}
apply
函数中做什么?也许你可以与我们分享相应的代码。 - Till Rohrmannif(last_event_ts == 0L){ last_event_ts = value.getTs(); } else { dwell_time += value.getTs() - last_event_ts; last_event_ts = value.getTs(); } } out.collect(new Features(tuple.getField(0), tuple.getField(2), tuple.getField(1), start_ts, end_ts, size, dwell_time, Boolean.FALSE));
- Vladimir Stoyak