CODEWITHSUNDEEP
pythongoogle-cloud-dataflowapache-beam
Lowblow
Lowblow

Reputation: 109

Correct syntax for defining custom trigger with OrFinally in Apache Beam in Python?

I am trying to define a custom trigger for a sliding window that triggers repeatedly for every element, but also triggers finally at the end of the watermark. I've looked around documentation for almost an hour now but have yet to find any example :(.

        | beam.WindowInto(
            beam.window.SlidingWindows(60, 10),
            trigger= Repeatedly(
                (AfterCount(1), OrFinally(AfterWatermark()))
            ),
            accumulation_mode=beam.transforms.trigger.AccumulationMode.DISCARDING
        )

This is what I'm trying right now and clearly this doesn't work but I am extremely lost in what the correct syntax is.

Upvotes: 1

Views: 591

Answers (1)

Jayadeep Jayaraman
Jayadeep Jayaraman

Reputation: 2825

Can you try changing the trigger like below and see

trigger=OrFinally(Repeatedly(AfterCount(1)), AfterWatermark()),
accumulation_mode=beam.transforms.trigger.AccumulationMode.DISCARDING

Upvotes: 3

Related Questions