CODEWITHSUNDEEP
google-cloud-dataflowapache-beam
mark
mark

Reputation: 73

Calculating periodic checkpoints for an unbounded stream in Apache Beam/DataFlow

I am using a global unbounded stream in combination with Stateful processing and timers in order to totally order a stream per key by event timestamp. The solution is described with the answer to this question:

Processing Total Ordering of Events By Key using Apache Beam

In order to restart the pipeline after a failure or stopping for some other reason, I need to determine the lowest event timestamp at which we are guaranteed that all other events have been processed downstream. This timestamp can be calculated periodically and persisted to a datastore and used as the input to the source IO (Kinesis) so that the stream can be re-read without having to go back to the beginning. (It is ok for us to have events replayed)

I considered having the stateful transformation emit the lowest processed timestamp as the output when the timer triggers and then combine all the outputs globally to find the minimum value. However, it is not possible to use a Global combine operation because a either a Window or a Trigger must be applied first.

Assuming that my stateful transform emits a Long when the timer fires which represents the smallest timestamp, I am defining the pipeline like this:

p.apply(events)
 .apply("StatefulTransform", ParDo.of(new StatefulTransform()))
 .apply(Window.<Long>configure().triggering(Repeatedly.forever(AfterFirst.of(
                    AfterPane.elementCountAtLeast(100),
                    AfterProcessingTime.pastFirstElementInPane().plusDelayOf(Duration.standardMinutes(1))))))
 .apply(Combine.globally(new MinLongFn()))
 .apply("WriteCheckpoint", ParDo.of(new WriteCheckpoint()));

Will this ensure that the checkpoints will only be written when all of the parallel workers have emitted at least one of their panes? I am concerned that a the combine operation may operate on panes from only some of the workers, e.g. there may be a worker that has either failed or is still waiting for another event to trigger it's timer.

Upvotes: 1

Views: 1475

Answers (1)

flyisland
flyisland

Reputation: 397

I'm a newbie of the Beam, but according to this blog https://beam.apache.org/blog/2017/08/16/splittable-do-fn.html, Splittable DoFn might be the thing you are looking for!

You could create an SDF to fetch the stream and accept the input element as the start point.

Upvotes: 0

Related Questions