apache-flink:count窗口超时

2vuwiymt  于 2021-06-24  发布在  Flink
关注(0)|答案(3)|浏览(636)

下面是一个简单的代码示例来说明我的问题：

  1. case class Record( key: String, value: Int )
  3. object Job extends App
  4. {
  5.   val env = StreamExecutionEnvironment.getExecutionEnvironment
  6.   val data = env.fromElements( Record("01",1), Record("02",2), Record("03",3), Record("04",4), Record("05",5) )
  7.   val step1 = data.filter( record => record.value % 3 != 0  ) // introduces some data loss
  8.   val step2 = data.map( r => Record( r.key, r.value * 2 ) )
  9.   val step3 = data.map( r => Record( r.key, r.value * 3 ) )
  10.   val merged = step1.union( step2, step3 )
  11.   val keyed = merged.keyBy(0)
  12.   val windowed = keyed.countWindow( 3 )
  13.   val summed = windowed.sum( 1 )
  14.   summed.print()
  15.   env.execute("test")
  16. }

这将产生以下结果：

  1. Record(01,6)
  2. Record(02,12)
  3. Record(04,24)
  4. Record(05,30)

正如预期的那样，不会为键“03”生成任何结果，因为计数窗口需要3个元素，而流中只有两个元素。
我想要的是某种带有超时的count窗口，这样，在某个超时之后，如果没有达到count窗口所期望的元素数，则使用现有元素生成部分结果。
对于这种行为，在我的示例中，当达到超时时将生成一条记录（03,15）。

scalaapache-flinkflink-streamingtimeout

3条答案

按热度按时间
92dk7w1h

92dk7w1h1#

我认为您可以使用processfunction实现这个用例
其中有count属性和windowend属性。使用它，您可以决定何时收集数据。

  1. public class TimeCountWindowProcessFunction extends ProcessFunction {
  3.     protected long windowStart;
  4.     protected long windowEnd;
  5.     protected long count;
  6.     private ValueState<CountPojo> state;
  8.     public TimeCountWindowProcessFunction(long windowSize, long count) {
  10.     this.windowSize = windowSize;
  11.     this.count = count;
  13.     }
  15. @Override
  16.     public void open(Configuration parameters) {
  18.     TypeInformation<CountPojo> typeInformation = TypeInformation.of(new TypeHint<CountPojo>() {
  19.     });
  20.     ValueStateDescriptor<CountPojo> descriptor = new ValueStateDescriptor("test", typeInformation);
  22.     state = getRuntimeContext().getState(descriptor);
  23. }
  25.     @Override
  26.     public void processElement(CountPojo input, Context ctx, Collector<CountPojo> out)
  27.             throws Exception {
  29.     long timestamp = ctx.timestamp();
  30.         windowStart = timestamp - (timestamp % windowSize);
  31.         windowEnd = windowStart + windowSize;
  33.         // retrieve the current count
  34.         CountPojo current = (CountPojo) state.value();
  36.         if (current == null) {
  38.             current = new CountPojo();
  39.         current.count = 1;
  41.             ctx.timerService().registerEventTimeTimer(windowEnd);
  42.         } else {
  44.             current.count += 1;
  45.         }
  47.         if(current.count >= count) {
  48.         out.collect(current);
  49.     }
  51.         // set the state's timestamp to the record's assigned event time timestamp
  52.         current.setLastModified(ctx.timestamp());
  54.         // write the state back
  55.         state.update(current);
  56.     }
  58.     @Override
  59.     public void onTimer(long timestamp, OnTimerContext ctx, Collector<CountPojo> out)
  60.             throws Exception {
  62.         if (windowEnd == timestamp) {
  64.             out.collect(state.value());
  65.         }
  67.         state.clear();
  68.     }
  69. }

我希望这对你有帮助。

展开查看全部
赞(0）分享  回复(0）举报  2021-06-24
e4yzc0pl

e4yzc0pl2#

你也可以用一个自定义窗口来实现 Trigger 它在达到计数或超时过期时激发，有效地混合了内置的 CountTrigger 以及 EventTimeTrigger .

赞(0）分享  回复(0）举报  2021-06-24
ubby3x7f

ubby3x7f3#

我遵循了大卫和尼拉夫的方法，下面是结果。
1） 使用自定义触发器：
在这里我颠倒了我最初的逻辑。我没有使用“计数窗口”，而是使用一个“时间窗口”，其持续时间与超时相对应，后跟一个触发器，在处理完所有元素后触发。

  1. case class Record( key: String, value: Int )
  3. object Job extends App
  4. {
  5.   val env = StreamExecutionEnvironment.getExecutionEnvironment
  6.   val data = env.fromElements( Record("01",1), Record("02",2), Record("03",3), Record("04",4), Record("05",5) )
  7.   val step1 = data.filter( record => record.value % 3 != 0  ) // introduces some data loss
  8.   val step2 = data.map( r => Record( r.key, r.value * 2 ) )
  9.   val step3 = data.map( r => Record( r.key, r.value * 3 ) )
  10.   val merged = step1.union( step2, step3 )
  11.   val keyed = merged.keyBy(0)
  12.   val windowed = keyed.timeWindow( Time.milliseconds( 50 ) )
  13.   val triggered = windowed.trigger( new CountTriggerWithTimeout( 3, env.getStreamTimeCharacteristic ) )
  14.   val summed = triggered.sum( 1 )
  15.   summed.print()
  16.   env.execute("test")
  17. }

下面是触发代码：

  1. import org.apache.flink.annotation.PublicEvolving
  2. import org.apache.flink.api.common.functions.ReduceFunction
  3. import org.apache.flink.api.common.functions.RuntimeContext
  4. import org.apache.flink.api.common.state.ReducingState
  5. import org.apache.flink.api.common.state.ReducingStateDescriptor
  6. import org.apache.flink.api.common.typeutils.base.LongSerializer
  7. import org.apache.flink.streaming.api.TimeCharacteristic
  8. import org.apache.flink.streaming.api.windowing.triggers._
  9. import org.apache.flink.streaming.api.windowing.triggers.Trigger.TriggerContext
  10. import org.apache.flink.streaming.api.windowing.windows.TimeWindow
  12. /**
  13.  * A trigger that fires when the count of elements in a pane reaches the given count or a 
  14.  * timeout is reached whatever happens first.
  15.  */
  16. class CountTriggerWithTimeout[W <: TimeWindow](maxCount: Long, timeCharacteristic: TimeCharacteristic) extends Trigger[Object,W] 
  17. {
  18.   private val countState: ReducingStateDescriptor[java.lang.Long] = new ReducingStateDescriptor[java.lang.Long]( "count", new Sum(), LongSerializer.INSTANCE)
  20.   override def onElement(element: Object, timestamp: Long, window: W, ctx: TriggerContext): TriggerResult = 
  21.   {
  22.       val count: ReducingState[java.lang.Long] = ctx.getPartitionedState(countState)
  23.       count.add( 1L )
  24.       if ( count.get >= maxCount || timestamp >= window.getEnd ) TriggerResult.FIRE_AND_PURGE else TriggerResult.CONTINUE
  25.   }
  27.   override def onProcessingTime(time: Long, window: W, ctx: TriggerContext): TriggerResult = 
  28.   {
  29.       if (timeCharacteristic == TimeCharacteristic.EventTime) TriggerResult.CONTINUE else
  30.       {
  31.           if ( time >= window.getEnd ) TriggerResult.CONTINUE else TriggerResult.FIRE_AND_PURGE
  32.       }
  33.   }
  35.   override def onEventTime(time: Long, window: W, ctx: TriggerContext): TriggerResult = 
  36.   {
  37.       if (timeCharacteristic == TimeCharacteristic.ProcessingTime) TriggerResult.CONTINUE else
  38.       {
  39.           if ( time >= window.getEnd ) TriggerResult.CONTINUE else TriggerResult.FIRE_AND_PURGE
  40.       }
  41.   }
  43.   override def clear(window: W, ctx: TriggerContext): Unit = 
  44.   {
  45.           ctx.getPartitionedState( countState ).clear
  46.     }
  48.     class Sum extends ReduceFunction[java.lang.Long] 
  49.   {
  50.         def reduce(value1: java.lang.Long, value2: java.lang.Long): java.lang.Long = value1 + value2
  51.   }
  52. }

2） 使用过程函数：

  1. case class Record( key: String, value: Int )
  3. object Job extends App
  4. {
  5.   val env = StreamExecutionEnvironment.getExecutionEnvironment
  6.   env.setStreamTimeCharacteristic( TimeCharacteristic.IngestionTime )
  7.   val data = env.fromElements( Record("01",1), Record("02",2), Record("03",3), Record("04",4), Record("05",5) )
  8.   val step1 = data.filter( record => record.value % 3 != 0  ) // introduces some data loss
  9.   val step2 = data.map( r => Record( r.key, r.value * 2 ) )
  10.   val step3 = data.map( r => Record( r.key, r.value * 3 ) )
  11.   val merged = step1.union( step2, step3 )
  12.   val keyed = merged.keyBy(0)
  13.   val processed = keyed.process( new TimeCountWindowProcessFunction( 3, 100 ) )
  14.   processed.print()
  15.   env.execute("test")
  16. }

当所有逻辑（即，窗口、触发和求和）进入函数时：

  1. import org.apache.flink.streaming.api.functions._
  2. import org.apache.flink.util._
  3. import org.apache.flink.api.common.state._
  5. case class Status( count: Int, key: String, value: Long )
  7. class TimeCountWindowProcessFunction( count: Long, windowSize: Long ) extends ProcessFunction[Record,Record] 
  8. {
  9.     lazy val state: ValueState[Status] = getRuntimeContext
  10.       .getState(new ValueStateDescriptor[Status]("state", classOf[Status]))
  12.     override def processElement( input: Record, ctx: ProcessFunction[Record,Record]#Context, out: Collector[Record] ): Unit =
  13.     {
  14.         val updated: Status = Option( state.value ) match {
  15.             case None => {
  16.                 ctx.timerService().registerEventTimeTimer( ctx.timestamp + windowSize )
  17.                 Status( 1, input.key, input.value )
  18.             }
  19.             case Some( current ) => Status( current.count + 1, input.key, input.value + current.value )    
  20.         }
  21.         if ( updated.count == count ) 
  22.         {
  23.             out.collect( Record( input.key, updated.value ) )
  24.             state.clear
  25.         }
  26.         else
  27.         {
  28.             state.update( updated )  
  29.         }        
  30.     }
  32.     override def onTimer( timestamp: Long, ctx: ProcessFunction[Record,Record]#OnTimerContext, out: Collector[Record] ): Unit =
  33.     {
  34.         Option( state.value ) match {
  35.             case None => // ignore
  36.             case Some( status ) => {
  37.                 out.collect( Record( status.key, status.value ) )
  38.                 state.clear
  39.             }
  40.         }
  41.     }
  42. }
展开查看全部
赞(0）分享  回复(0）举报  2021-06-24
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com