我实现了一个数据管道,该管道具有多个无限制的源和边输入,将数据与滑动窗口(每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中的解释很好

“由于多个窗口重叠,所以数据集中的大多数元素将属于一个以上的窗口。这种窗口化对于获取数据的运行平均值很有用;……”

固定的窗口不会重叠:

“固定的时间窗口表示数据流中一致的持续时间,不重叠的时间间隔。”

09-10 09:00
查看更多