如何在Apache beam python中创建任务之间的依赖关系

Question

import apache_beam as beam
import logging
from apache_beam.options.pipeline_options import PipelineOptions

class Sample(beam.PTransform):
    def __init__(self, index):
        self.index = index

    def expand(self, pcoll):
        logging.info(self.index)
        return pcoll

class LoadData(beam.DoFn):
    def process(self, context):
        logging.info("***")

if __name__ == '__main__':

    logging.getLogger().setLevel(logging.INFO)
    pipeline = beam.Pipeline(options=PipelineOptions())

    (pipeline
        | "one" >> Sample(1)
        | "two: Read" >> beam.io.ReadFromText('sample.json')
        | "three: show" >> beam.ParDo(LoadData())
        | "four: sample2" >> Sample(2)
    )
    pipeline.run().wait_until_finish()

我希望它会按照一、二、三、四的顺序执行。但它是在并行模式下运行的。

以上代码的输出：

INFO:root:Missing pipeline option (runner). Executing pipeline using the 
default runner: DirectRunner.
INFO:root:1
INFO:root:2
INFO:root:Running pipeline with DirectRunner.
INFO:root:***
INFO:root:***
INFO:root:***
INFO:root:***
INFO:root:***

Answer 1

根据Dataflow的文档：

当管道运行者为分布式执行构建实际管道时，管道可能会被优化。例如，将某些转换一起运行或以不同的顺序运行可能在计算上更有效。Dataflow服务完全管理管道执行的这一方面。

同样根据Apache Beam的文档：