我实现了一个数据管道,该管道具有多个无限制的源和边输入,将数据与滑动窗口(每30s和每10s)结合在一起,并将转换后的输出发送到Kafka主题中。我的问题是,在窗口的前10秒接收到的数据会发射3次(即)每当一个新窗口开始直到第一个窗口完成时才触发。如何只发出一次转换的数据或避免重复?
我已经使用了废弃触发窗格,并且没有任何区别。每当我尝试将“关闭窗口”行为设置为FIRE_ALWAYS / FIRE_IF_NON_EMPTY时,都会引发以下错误。
线程“主”中的异常org.apache.beam.sdk.Pipeline $ PipelineExecutionException:java.lang.IllegalArgumentException:空PCollection作为单例视图访问。考虑设置withDefault以提供默认值
在org.apache.beam.runners.direct.DirectRunner $ DirectPipelineResult.waitUntilFinish(DirectRunner.java:332)
在org.apache.beam.runners.direct.DirectRunner $ DirectPipelineResult.waitUntilFinish(DirectRunner.java:302)
在org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:197)
在org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:64)
在org.apache.beam.sdk.Pipeline.run(Pipeline.java:313)
在org.apache.beam.sdk.Pipeline.run(Pipeline.java:299)
在y.yyy.main(yyy.java:86)
原因:java.lang.IllegalArgumentException:空PCollection以单例视图访问。考虑设置withDefault以提供默认值
在org.apache.beam.sdk.transforms.View $ SingletonCombineFn.identity(View.java:378)
在org.apache.beam.sdk.transforms.Combine $ BinaryCombineFn.extractOutput(Combine.java:481)
在org.apache.beam.sdk.transforms.Combine $ BinaryCombineFn.extractOutput(Combine.java:429)
在org.apache.beam.sdk.transforms.Combine $ CombineFn.apply(Combine.java:387)
在org.apache.beam.sdk.transforms.Combine $ GroupedValues $ 1.processElement(Combine.java:2089)
data.apply("Transform", ParDo.of(
new DoFn<String, Row>() {
private static final long serialVersionUID = 1L;
@ProcessElement
public void processElement(
ProcessContext processContext,
final OutputReceiver<Row> emitter) {
String record = processContext.element();
final String[] parts = record.split(",");
emitter.output(Row.withSchema(sch).addValues(parts).build());
}
})).apply(
"window1",
Window
.<Row>into(
SlidingWindows
.of(Duration.standardSeconds(30))
.every(Duration.standardSeconds(10)))
.withAllowedLateness(
Duration.ZERO,
Window.ClosingBehavior.FIRE_IF_NON_EMPTY)
.discardingFiredPanes());
请引导我仅触发一次窗口(即,我不想发送已处理的记录)
更新:侧面输入的上述错误经常发生,并且不是由于Windows引起的,这似乎是Apache Beam(https://issues.apache.org/jira/browse/BEAM-6086)中的问题
我尝试使用State来识别行是否已被处理,但是该状态未保留或未设置。 (即)读取状态时,我总是会得到null。
public class CheckState extends DoFn<KV<String,String>,KV<Integer,String>> {
private static final long serialVersionUID = 1L;
@StateId("count")
private final StateSpec<ValueState<String>> countState =
StateSpecs.value(StringUtf8Coder.of());
@ProcessElement
public void processElement(
ProcessContext processContext,
@StateId("count") ValueState<String> countState) {
KV<String,String> record = processContext.element();
String row = record.getValue();
System.out.println("State: " + countState.read());
System.out.println("Setting state as "+ record.getKey() + " for value"+ row.split(",")[0]);
processContext.output(KV.of(current, row));
countState.write(record.getKey());
}
谢谢,
最佳答案
如果我正确理解了这个问题,则可能与管道中滑动窗口的使用有关:
滑动时间窗口重叠,光束导引Window Functions中的解释很好
“由于多个窗口重叠,所以数据集中的大多数元素将属于一个以上的窗口。这种窗口化对于获取数据的运行平均值很有用;……”
固定的窗口不会重叠:
“固定的时间窗口表示数据流中一致的持续时间,不重叠的时间间隔。”