Trigger Beam ParDo at window closing only

user_1077748 0
Viewed 127

I have a pipeline that read events from Kafka. I want to count and log the event count only when the window closes. By doing this I will only have one output log per Kafka partition/shard on each window. I use a timestamp in the header which I truncate to the hour to create a collection of hourly timestamps. I group the timestamps by hour and I log the hourly timestamp and count. This log will be sent to Grafana to create a dashboard with the counts.

Below is how I fetch the data from Kafka and where it defines the window duration:

int windowDuration = 5;
p.apply("Read from Kafka",KafkaIO.<byte[], GenericRecord>read()
            .withBootstrapServers(options.getSourceBrokers().get())
            .withTopics(srcTopics)
            .withKeyDeserializer(ByteArrayDeserializer.class)
            .withValueDeserializer(ConfluentSchemaRegistryDeserializerProvider
            .of(options.getSchemaRegistryUrl().get(), options.getSubject().get()))
                    .commitOffsetsInFinalize())
  .apply("Windowing of " + windowDuration +" seconds" , 
            Window.<KafkaRecord<byte[], GenericRecord>>into(
            FixedWindows.of(Duration.standardSeconds(windowDuration))));

The next step in the pipeline is to produce two collections from the above collection one with the events as GenericRecord and the other with the hourly timestamp, see below. I want a trigger (I believe) to be applied only two the collection holding the counts. So that it only prints the count once per window. Currently as is, it prints a count every time it reads from Kafka creating a large number of entries.

  tuplePCollection.get(createdOnTupleTag)
  .apply(Count.perElement())
  .apply( MapElements.into(TypeDescriptors.strings())
  .via( (KV<Long,Long> recordCount) -> recordCount.getKey() +
    ": " + recordCount.getValue()))
  .apply( ParDo.of(new LoggerFn.logRecords<String>()));

Here is the DoFn I use to log the counts:

 class LoggerFn<T> extends DoFn<T, T> {
        @ProcessElement
        public void process(ProcessContext c) {
            T e = (T)c.element();
            LOGGER.info(e);
            c.output(e);
        }
    }
apache-beam
1 Answers

You can use the trigger “Window.ClosingBehavior”. You need to specify under which conditions a final pane will be created when a window is permanently closed. You can use these options:

  • FIRE_ALWAYS: Always Fire the last Pane.

  • FIRE_IF_NON_EMPTY: Only Fire the last pane if there is new data since previous firing.

You can see this example.

// We first specify to never emit any panes
 .triggering(Never.ever())
 
 // We then specify to fire always when closing the window. This will emit a
 // single final pane at the end of allowedLateness
 .withAllowedLateness(allowedLateness, Window.ClosingBehavior.FIRE_ALWAYS)
 .discardingFiredPanes())

You can see more information about this trigger.

user_169291450
Related