Composer DAG 触发 Dataflow 作业并成功运行，但最终文件未出现在输出桶中

Question

我已经设置了一个运行数据流作业的 DAG。 Dag 触发它正常，并且它运行成功，但输出文件没有出现在输出位置。输出位置是另一个项目中的存储桶，并且正在使用的 SA 有权写入该存储桶...知道为什么该文件没有生成吗？

DF 作业：

import apache_beam as beam
from apache_beam.options.value_provider import StaticValueProvider
from apache_beam.options.pipeline_options import PipelineOptions
from datetime import datetime
import logging


class UserOptions(PipelineOptions):
        @classmethod
        def _add_argparse_args(cls, parser):
                parser.add_value_provider_argument('--templated_int', type=int)
                parser.add_value_provider_argument("--input", type=str )
                parser.add_value_provider_argument("--output", type=str )

class process_file(beam.DoFn):
        def __init__(self, templated_int):
                self.templated_int = templated_int

        def process(self, an_int):
                yield self.templated_int.get() + an_int

def clean_file():
        pipeline_options = PipelineOptions()
        user_options = pipeline_options.view_as(UserOptions)
        tstmp = datetime.now().strftime("%Y%m%d%H")
        output = user_options.output 
        logging.info('Input: ', user_options.input)
        logging.info('Output: ', output)
                                
        with beam.Pipeline(options=pipeline_options) as p:

                p | 'Read from a File' >> beam.io.ReadFromText(user_options.input, skip_header_lines=1) | 'Split into rows' >> beam.Map(lambda x:x.split(",")) | 'Confirm index locations' >> beam.Map(lambda x:f'{x[0]},{x[1]}{x[2]}{x[3]}{x[4]},{x[5]}') | 'Write to clean file' >> beam.io.WriteToText(output) 
        p.run().wait_until_finish()

if __name__ == "__main__":
        clean_file()

Answer 1

当您在 Dataflow Pipeline 图表中选择一个步骤时，日志面板会切换显示由 Dataflow 服务生成的作业日志，显示来自运行管道步骤的 Compute Engine 实例的日志。

Cloud Logging 将从项目的 Compute Engine 实例收集的所有日志合并到一个位置。此外，请参阅记录管道消息，了解有关使用 dataflow 的各种日志记录的详细信息能力。