Flink的timeWindow of DataStream有问题[寻求帮助]

Question

我是Flink的初学者。我可以实现Flink流。我还练习了关于KeyedStream的timeWindow。现在我想使用Flink关于DataStream的timeWindow来实现批处理操作。我尝试了一些代码，但它就是不能工作。我不知道怎样才能达到预期的效果。

val resultDataStream: DataStream[HBaseOperation] = env
  .addSource(kafkaUserEventSource)
  .assignTimestampsAndWatermarks(new Launcher.CustomWatermarkExtractor(Time.hours(24)))
  .flatMap { line =>
    import scala.collection.JavaConverters._
    val tableInfo: mutable.Buffer[TableInfo] = JSON.parseArray(line.columnValueList, classOf[TableInfo]).asScala
    line.eventType match {
      case EventType.INSERT => tableInfo.map {
        row => HBaseOperation(line.eventType, s"${line.dbName}.${line.tableName}", HBASE_CFNAME, tableInfo(0).columnValue, row.columnName, row.columnValue)
      }
      case EventType.UPDATE => tableInfo.filter(_.isValid).map {
        row => HBaseOperation(line.eventType, s"${line.dbName}.${line.tableName}", HBASE_CFNAME, tableInfo(0).columnValue, row.columnName, row.columnValue)
      }
      case EventType.DELETE => List(HBaseOperation(line.eventType, s"${line.dbName}.${line.tableName}", HBASE_CFNAME, tableInfo(0).columnValue, null, null))
    }
  }
resultDataStream.print()//Here it works

//batch operations
val value: AllWindowedStream[HBaseOperation, TimeWindow] = resultDataStream.timeWindowAll(Time.seconds(5))
val resultDataStream2: DataStream[BatchHBaseOperation] = value.apply(new RichAllWindowFunction[HBaseOperation, BatchHBaseOperation, TimeWindow] {
  override def apply(window: TimeWindow, input: Iterable[HBaseOperation], out: Collector[BatchHBaseOperation]): Unit = {
    val ops: ListBuffer[HBaseOperation] = ListBuffer[HBaseOperation]()
    input.foreach(op => ops += op)
    out.collect(BatchHBaseOperation(ops.toList))
  }
})//Doesn't work
resultDataStream2.print()
//resultDataStream2.addSink(HBaseUtil.putMapData(_))
env.execute("KafkaHBaseApp")

Answer 1

你的水印生成器的细节还不清楚，但如果它是基于Flink的有界无序水印，那么CustomWatermarkExtractor(Time.hours(24))意味着在24小时(加5秒)的数据处理完成之前，时间窗口不会第一次被触发。这可以解释为什么它看起来不起作用。