Add dependency between 2 Dofn in Apache Beam

Question

Is there any way I can create dependency between 2 Dofn, so that it will wait for 1st Dofn method to complete then second Dofn method will run. Just wondering how we can achieve this use-case.

I meant 2nd dofn should wait until the completion of 1st Dofn. — user12878180
By "completion" you mean run over all the elements before the 2nd runs over any of them? If so, the answer below is good. — robertwb

Alex Alex · Accepted Answer · 2020-02-14T18:54:48

There may be a cleaner way to do this, but I've noticed that doing the following will acheive the effect you desire:

Route the output of the first DoFn to also go to a counter, and then have the output of that counter be passed into the ParDo of the second DoFn as a side input

class DoFn2(apache_beam.DoFn):
    def process(self, element, count_do_fn_1_output, *args, **kwargs):
        # ...

do_fn_1_output = do_fn_1_input | 'do fn 1' >> apache_beam.ParDo(DoFn1())

count_do_fn_1_output = (
    do_fn_1_output 
    | 'count do_fn_1_output' >> apache_beam.combiners.Count.Globally())

do_fn_2_output = (
    do_fn_1_output 
    | 'do fn 2' >> apache_beam.ParDo(DoFn2(), count_do_fn_1_output=apache_beam.pvalue.AsSingleton(count_do_fn_1_output)))

Add dependency between 2 Dofn in Apache Beam

2 Answers