在 GCP 数据流上运行脚本答案

【问题标题】：Run script on GCP Dataflow在 GCP 数据流上运行脚本
【发布时间】：2020-06-09 11:07:43
【问题描述】：

我开始尝试 Google Cloud Dataflow，在经典的wordcount 示例之后，我编写了自己的脚本：

import argparse
import sys

import apache_beam as beam
from apache_beam.io import ReadFromText
from apache_beam.io import WriteToText
from apache_beam.options.pipeline_options import PipelineOptions


class Split(beam.DoFn):

    def process(self, element):
        (numfact, bag, type, owner, 
         main_owner, client) = element.splt('\t')

        return [{
            'numfact': int(numfact),
            'type': type,
            'owner': owner
        }]


parser = argparse.ArgumentParser()

parser.add_argument('--input')
parser.add_argument('--output')

known_args, extra_args = parser.parse_known_args(sys.argv[1:])

options = PipelineOptions(extra_args)
p = beam.Pipeline(options=options)
print(known_args)
print(extra_args)
csv_lines = (p | "Load" >> ReadFromText(known_args.input, skip_header_lines=1) | "Process" >> beam.ParDo(Split()) | "Write" >> WriteToText(known_args.output))

这里是输入文件的示例：

Numfact BAG TYPE    OWNER   MAIN OWNER  CLIENT
728632636   CNT Alternativos    Kramer Ortiz    ACCIDENTES PERSONALES TELETICKET    Rimac
704845964   CNT Alternativos    Kramer Ortiz    SOAT    Canal
701387639   CNT SIN ASIGNAR Sin asignar WEB VEHICULOS   Canal
692571746   CNT Concesionarios  Kramer Ortiz    WEB VEHICULOS   Canal
682823453   CNT Alternativos    Kramer Ortiz    WEB VEHICULOS   Canal
682823452   CNT Alternativos    Kramer Ortiz    WEB VEHICULOS   Canal
682823451   CNT Alternativos    Kramer Ortiz    WEB VEHICULOS   Canal
682823454   CNT Alternativos    Kramer Ortiz    WEB VEHICULOS   Canal
706853395   CNT Alternativos    Kramer Ortiz    ACCIDENTES PERSONALES - WEB Canal
706466281   CNT Alternativos    Kramer Ortiz    SOAT    Canal

最后，我这样调用它来执行（文件保存为.txt）：

python -m beam --input gs://dummy_bucket/data_entry/pcd/pcd_ensure.txt --output gs://dummy_bucket/outputs --runner DataflowRunner --project dummyproject-268120 --temp_location gs://dummy_bucket/tmp --region us-central1

在此之后，它会在控制台上显示打印，但不会在 DataFlow 控制台中注册执行。

更新

这是控制台的样子：

(gcp) gocht@~/script$ python -m beam --input gs://dummy_bucket/data_entry/pcd/pcd_ensure.txt --output gs://dummy_bucket/outputs --runner DataflowRunner --project dummyproject-268120 --temp_location gs://dummy_bucket/tmp --region us-central1
Namespace(input='gs://dummy_bucket/data_entry/pcd/pcd_ensure.txt', output='gs://dummy_bucket/outputs')   ['--runner', 'DataflowRunner', '--project', 'dummyproject-268120', '--temp_location', 'gs://dummy_bucket/tmp', '--region', 'us-central1']

这仅显示放置在代码脚本上的打印。

我错过了什么？

谢谢！

【问题讨论】：

你能提供命令的输出吗？
@Pievis 已更新。
我第一次看到 python 版本但是代码中可能缺少 p.run().wait_until_finish() 吗？
@Pievis 工作，伙计！如此简单，如此有效。考虑将其发布为答案。谢谢！
很高兴能帮上忙！我正在听从您的建议并将其作为答案:)

标签： python google-cloud-platform google-cloud-dataflow pipeline apache-beam

【解决方案1】：

既然答案在评论里，也写在这里:)

您需要通过执行以下操作来实际运行管道：

p.run().wait_until_finish()

如果您觉得卡住并且不确定哪里出了问题，请尝试查看here 提供的示例 - java 版本在开始使用数据流时真的帮助了我很多:)

【讨论】：

【解决方案2】：

你需要

result = p.run()

在文件末尾运行管道。

基本上我认为你已经构建了你的管道，但并没有真正要求运行它。

【讨论】：