Flink会话窗口未按预期工作

Question

Flink中的会话窗口在prod env上没有按预期工作（相同的逻辑在本地env上工作）。这个想法是为特定的用户ID发出“sample_event_two”的计数

  public void process(
      String s,
      Context context,
      Iterable<SampleEvent> sampleEvents,
      Collector<EnrichedSampleEvent> collector)
      throws Exception {

    EnrichedSampleEvent event = null;
    boolean isSampleEventOnePresent = false;
    int count = 0;


    for (SampleEvent sampleEvent : sampleEvents) {

      if (sampleEvent.getEventName().equals("sample_event_one_name")) {

        Logger.info("Received sample_event_one for userId: {}");
        isSampleEventOnePresent = true;

      } else {
        // Calculate the count for sample_event_two
        count++;

        if (Objects.isNull(event)) {
          event = new EnrichedSampleEvent();
          event.setUserId(sampleEvent.getUserId());
        }
      }
    }

    if (isSampleEventOnePresent && Objects.nonNull(event)) {
      Logger.info(
          "Created EnrichedSampleEvent for userId: {} with count: {}",
          event.getUserId(),
          event.getCount());
      collector.collect(event);
    } else if (Objects.nonNull(event)) {
      Logger.info(
          "No sampleOneEvent event found sampleTwoEvent with userId: {}, count: {}",
          event.getUserId(),
          count);
    }
  }

尽管集合中存在sample_event_one（通过验证日志消息“已接收sample_event_one”是否存在来确认）并且计数计算正确，但我没有看到任何输出事件被创建。我看到日志消息“未找到 sampleOneEvent 事件，而不是发出 EnrichedSampleEvent，而是看到 userID：”123，count： 5”。有人可以帮我解决这个问题吗？

Answer 1

您的ProcessWindowFunction将分别为每个键调用。由于密钥是用户id和记录id的组合，因此仅知道“Received sample_event_one”出现在同一用户的日志中是不够的。即使是同一个用户，也可能有不同的记录id。