未触发flink翻滚窗口（无水印策略）

Question

问题陈述：来自kafka源的流式事件。这些事件有效载荷为字符串格式。将它们解析为文档，并根据事件时间每隔5秒将其批量插入DB。

map（）函数正在执行。但程序控制不会进入Application（）。因此不会发生批量插入。我尝试了键控和非键控窗口。它们都不工作。没有抛出错误。

flink版本：1.15.0

下面是我的主要方法的代码。我应该如何解决这个问题？

  public static void main(String[] args) throws Exception {

    final Logger logger = LoggerFactory.getLogger(Main.class);
    final StreamExecutionEnvironment streamExecutionEnv = StreamExecutionEnvironment.getExecutionEnvironment();
    streamExecutionEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);


    KafkaConfig kafkaConfig = Utils.getAppConfig().getKafkaConfig();

    logger.info("main() Loading kafka config :: {}", kafkaConfig);

    KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
        .setBootstrapServers(kafkaConfig.getBootstrapServers())
        .setTopics(kafkaConfig.getTopics())
        .setGroupId(kafkaConfig.getConsumerGroupId())
        .setStartingOffsets(OffsetsInitializer.latest())
        .setValueOnlyDeserializer(new SimpleStringSchema()).build();

    logger.info("main() Configured kafka source :: {}", kafkaSource);

    DataStreamSource<String> dataStream = streamExecutionEnv.fromSource(kafkaSource,
        WatermarkStrategy.noWatermarks(), "mySource");

    logger.info("main() Configured kafka dataStream :: {}", dataStream);

    DataStream<Document> dataStream1 = dataStream.map(new DocumentMapperFunction());

    DataStream<InsertOneModel<Document>> dataStream2 = dataStream1.map(new InsertOneModelMapperFunction());

    DataStream<Object> dataStream3 = dataStream2
        .windowAll(TumblingEventTimeWindows.of(Time.seconds(5), Time.seconds(0)))
            /*.keyBy(insertOneModel -> insertOneModel.getDocument().get("ackSubSystem"))
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))*/
        .apply(new BulkInsertToDB2())
            .setParallelism(1);

    logger.info("main() before streamExecutionEnv execution");

    dataStream3.print();

    streamExecutionEnv.execute();
  }

Answer 1

事件时间窗口需要水印策略。如果没有，窗口将永远不会触发。