아파치 빔 윈도우 - 고정 윈도우 닫지 않는 것 같습니까?

Question

Apache 빔 파이프 라인에서 고정 창을 사용하려고합니다 (DirectRunner 사용). 다음과 같이 우리의 흐름은 다음과 같습니다 하위 술집에서

1. 풀 데이터를/
, 각각의 창을 결합하여 직렬화 JSON 자바 객체로
2. 윈도우 이벤트 사용자 정의 CombineFn을 사용하여 오초
3. 의 고정 창 승/
4. 테스트를 위하여를 들어 List<Event>
5. 에 Event S, 단순히 출력 List<Event>

을 수득

---

파이프 라인 코드 :

pipeline 
       // Read from pubsub topic to create unbounded PCollection 
       .apply(PubsubIO 
        .<String>read() 
        .topic(options.getTopic()) 
        .withCoder(StringUtf8Coder.of()) 
       ) 

       // Deserialize JSON into Event object 
       .apply("ParseEvent", ParDo 
        .of(new ParseEventFn()) 
       ) 

       // Window events with a fixed window size of 5 seconds 
       .apply("Window", Window 
        .<Event>into(FixedWindows 
         .of(Duration.standardSeconds(5)) 
        ) 
       ) 

       // Group events by window 
       .apply("CombineEvents", Combine 
        .globally(new CombineEventsFn()) 
        .withoutDefaults() 
       ) 

       // Log grouped events 
       .apply("LogEvent", ParDo 
        .of(new LogEventFn()) 
       ); 

---

우리가보고있는 결과는 우리가 어떤 로깅을하지 않는 한 최종 단계는 실행되지 않습니다 것입니다.

또한 사용자 정의 CombineFn 클래스의 각 메소드에 System.out.println("***")을 추가하여 실행시기를 추적하고 실행되지 않는 것으로 보입니다.

여기서 창 설정이 잘못 되었습니까? 우리는 https://beam.apache.org/documentation/programming-guide/#windowing에있는 예제를 따라 갔고 상당히 단순 해 보였지만 분명히 빠진 것이 있습니다.

모든 통찰력을 환영합니다 - 미리 감사드립니다!

Answer 1

주요 문제가 실제로 실종 트리거였던 것처럼 보입니다 - 창문이 열리고 결과를 내 보내야 할 때 아무 것도 표시되지 않았습니다. 우리는 처리 시간 (없는 이벤트 시간) 그래서 다음 않았다 기반으로 간단하게 윈도우에 원 :

.apply("Window", Window 
    .<Event>into(new GlobalWindows()) 
    .triggering(Repeatedly 
     .forever(AfterProcessingTime 
      .pastFirstElementInPane() 
      .plusDelayOf(Duration.standardSeconds(5)) 
     ) 
    ) 
    .withAllowedLateness(Duration.ZERO).discardingFiredPanes() 
) 

는 기본적으로이 첫 번째 요소가 처리 된 후 이벤트를 5 초 방출하는 트리거 글로벌 창을 만듭니다. 창을 닫을 때마다 다른 요소가 수신되면 다른 창이 열립니다. 우리가 withAllowedLateness 조각을 가지고 있지 않았을 때 빔은 불평했다 - 내가 알고있는 한, 이것은 단지 늦은 데이터를 무시하도록 말한다.

여기에 약간의 이해가있을 수 있지만 위의 스 니펫이 문제를 해결했습니다.