How to implement a WindowableTask(similar to samza) in Apache flink?

[tuk]

I am new to flink and this is my first post in the community.

Samza has a concept of windowing where a stream processing job needs to do something in regular intervals, regardless of how many incoming messages the job is processing.

For example, a simple per-minute event counter in samza will be like below:

public class EventCounterTask implements StreamTask, WindowableTask {

  public static final SystemStream OUTPUT_STREAM =
    new SystemStream("kafka", "events-per-minute");

  private int eventsSeen = 0;

  public void process(IncomingMessageEnvelope envelope,
                      MessageCollector collector,
                      TaskCoordinator coordinator) {
    eventsSeen++;
  }

  public void window(MessageCollector collector,
                     TaskCoordinator coordinator) {
    collector.send(new OutgoingMessageEnvelope(OUTPUT_STREAM, eventsSeen));
    eventsSeen = 0;
  }
}

Can someone let me know how to implement an equivalent thing in apache flink (samza is single threaded so window and process will not happen concurrently) or point me to the relevant documentation?

[David Anderson-4]

Please note that I responded to this question on Stack Overflow: https://stackoverflow.com/questions/65414125/how-to-implement-a-windowabletask-similar-to-samza-in-apache-flink

Regards,

David

On Wed, Dec 23, 2020 at 7:08 AM Debraj Manna <[hidden email]> wrote:

I am new to flink and this is my first post in the community.

Samza has a concept of windowing where a stream processing job needs to do something in regular intervals, regardless of how many incoming messages the job is processing.

For example, a simple per-minute event counter in samza will be like below:

public class EventCounterTask implements StreamTask, WindowableTask {

  public static final SystemStream OUTPUT_STREAM =
    new SystemStream("kafka", "events-per-minute");

  private int eventsSeen = 0;

  public void process(IncomingMessageEnvelope envelope,
                      MessageCollector collector,
                      TaskCoordinator coordinator) {
    eventsSeen++;
  }

  public void window(MessageCollector collector,
                     TaskCoordinator coordinator) {
    collector.send(new OutgoingMessageEnvelope(OUTPUT_STREAM, eventsSeen));
    eventsSeen = 0;
  }
}

Can someone let me know how to implement an equivalent thing in apache flink (samza is single threaded so window and process will not happen concurrently) or point me to the relevant documentation?