状态和计时器的问题-apache beam

Question

我试图重新创建本博客文章中描述的apache beam管道的一个简单示例，该示例使用了状态和计时器。

以下是从博客中复制粘贴的Enrich DoFn：

public class Enrich extends DoFn<KV<String, String>, String> {
  private static final long serialVersionUID = 1L;

  private static final int MAX_BUFFER_SIZE = 2;

  @StateId("buffer")
  private final StateSpec<BagState<String>> bufferedEvents =
    StateSpecs.bag();

  @StateId("count")
  private final StateSpec<ValueState<Integer>> countState =
    StateSpecs.value();

  @TimerId("expiry")
  private final TimerSpec expirySpec = 
    TimerSpecs.timer(TimeDomain.EVENT_TIME);

  @ProcessElement
  public void process(
    ProcessContext context,
    BoundedWindow window,
    @StateId("buffer") BagState<String> bufferState,
    @StateId("count") ValueState<Integer> countState,
    @TimerId("expiry") Timer expiryTimer) {

      Duration allowedLateness = Duration.standardSeconds(10);

      expiryTimer.set(window.maxTimestamp().plus(allowedLateness));

      int count = firstNonNull(countState.read(), 0);
      count = count + 1;
      countState.write(count);
      bufferState.add(context.element().getValue());

      if (count >= MAX_BUFFER_SIZE) {
        for (String event : bufferState.read()) {
          context.output(enrichEvent(event));
        }
        bufferState.clear();
        countState.clear();
      }
    }

    @OnTimer("expiry")
    public void onExpiry(
      OnTimerContext context,
      @StateId("buffer") BagState<String> bufferState) {

        if (!bufferState.isEmpty().read()) {
          for (String event : bufferState.read()) {
            context.output(enrichEvent(event));
          }
          bufferState.clear();
        }
    }

    public static String enrichEvent(String event) {
      return event + ": enriched";
    }

    public static int firstNonNull(Integer x, Integer y) {
      if (x == null) {
        return y;
      }
      return x;
    }
}

下面是我用来测试enrichdofn:

@RunWith(JUnit4.class)
public class EnrichTest {
  final Logger LOG = LoggerFactory.getLogger(EnrichTest.class);

  @Rule
  public TestPipeline p = TestPipeline.create();

  static final String record1 = "1";
  static final String record2 = "2";
  static final String record3 = "3";

  static final String key = "a key";

  static final String result1 = "1: enriched";
  static final String result2 = "2: enriched";
  static final String result3 = "3: enriched";

  @Test
  public void testSimple() throws Exception {
    Duration ALLOWED_LATENESS = Duration.standardSeconds(10);
    Duration WINDOW_DURATION = Duration.standardSeconds(10);
    Instant baseTime = new Instant(0L);
    KvCoder<String, String> coder = 
      KvCoder.of(AvroCoder.of(String.class), AvroCoder.of(String.class));

    TestStream<KV<String, String>> items = 
        TestStream
          .create(coder)
          .advanceWatermarkTo(baseTime)
          .addElements(
              TimestampedValue.of(
                KV.of(key, record1),
                baseTime.plus(Duration.standardSeconds(1))))
          .addElements(
              TimestampedValue.of(
                KV.of(key, record2),
                baseTime.plus(Duration.standardSeconds(0))))
          .advanceWatermarkTo(
              baseTime.plus(Duration.standardSeconds(11)))
          .addElements(
              TimestampedValue.of(
                KV.of(key, record3),
                baseTime.plus(Duration.standardSeconds(2))))
          .advanceWatermarkToInfinity();

    PCollection<String> results = 
        p.apply(items)
         .apply(new CreateWindows (WINDOW_DURATION, ALLOWED_LATENESS))
         .apply(ParDo.of(new Enrich()));

    PAssert
      .that(results)
      .inWindow(new IntervalWindow(baseTime, WINDOW_DURATION))
      .containsInAnyOrder(result1, result2, result3);

    p.run().waitUntilFinish();
  }
}

下面是我的窗口函数：

public class CreateWindows extends 
  PTransform<PCollection<KV<String, String>>,
             PCollection<KV<String, String>>> {

  private static final long serialVersionUID = 1L;
  private final Duration windowDuration;
  private final Duration allowedLateness;

  public CreateStringWindows(Duration windowDuration, Duration allowedLateness) {
    this.windowDuration = windowDuration;
    this.allowedLateness = allowedLateness;
  }

  @Override
  public PCollection<KV<String, String>> expand(
    PCollection<KV<String, String>> items) {

    return items.apply("Aggregate fixed window",
      Window.<KV<String, String>>into(FixedWindows.of(windowDuration))
            .triggering(AfterWatermark.pastEndOfWindow())
            .discardingFiredPanes()
            .withAllowedLateness(allowedLateness));
  }
}

java.lang.IllegalStateException: TimestampCombiner moved element from 1970-01-01T00:00:19.999Z to earlier time 1970-01-01T00:00:09.999Z for window [1970-01-01T00:00:00.000Z..1970-01-01T00:00:10.000Z)

博客文章没有具体提到它使用的窗口化策略。这会是问题所在吗？我还尝试使用never.ever()作为窗口触发器，但得到相同的错误：

.triggering(Never.ever())
.discardingFiredPanes()
.withAllowedLateness(allowedLateness));

很抱歉这篇文章太长了，任何帮助都将不胜感激。

Answer 1

最后，我编辑了onexpiry以使用context.outputWithTimestamp（encryEvent（event）,window.maxTimestamp（））；而不是context.output（encryEvent（event））；。这就解决了问题。

下面是更正的onexpiry方法。

    @OnTimer("expiry")
    public void onExpiry(
      OnTimerContext context, BoundedWindow window,
      @StateId("buffer") BagState<String> bufferState) {

        if (!bufferState.isEmpty().read()) {
          for (String event : bufferState.read()) {
            context.outputWithTimestamp(enrichEvent(event), window.maxTimestamp());
          }
          bufferState.clear();
        }
    }