【发布时间】：2022-12-20 02:15:23
【问题描述】：
我正在开发一个数据流流作业，用于通过在 Cloud Storage 中创建对象（通过 Pub\Sub 通知）触发的 CSV 检查。
我正在使用数据流，因为这是业务需求和消息重复数据删除管理（可以通过 Pub\Sub 实现）。
在每个管道步骤中，我都会进行特定类型的控制（检查规则在我阅读的 Google 表格中定义，其中包含我在管道中创建的步骤）。 如果所有步骤都正常，则将文件复制到另一个存储桶中，否则发送错误邮件。由于这些原因，我需要一个全局函数来可能在所有步骤中调用。
我在库调用后声明了函数：
from email import header
from hashlib import new
from msilib.schema import Error
import json
import apache_beam as beam
from apache_beam import pvalue, window, GroupByKey
from datetime import datetime
import logging, uuid
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions
TOPIC = "TOPIC_PATH"
def test():
# send an email
print("Send email")
class ReadPubSubMessage(beam.DoFn):
def __init__(self):
self.prod_bucket = "prd-v2"
self.invalid_bucket = "prd-error"
def process(self, element, *args, **kwargs):
import uuid, json
from datetime import datetime
# Creating a uuid for the ingested file identification
try:
uuidOne = uuid.uuid1()
logging.info("Reading PubSub message")
# Reading the PubSub json end extracting main information
res = json.loads(element)
path_loaded_blob = res["id"]
type_object = res["contentType"]
# Getting the date from the blob path
list_of_path_parts = path_loaded_blob.split("/")
. # other code
.
.
yield final_list
except Exception as e:
test(email)
logging.error("Error: " + str(e))
beam_options = PipelineOptions()
google_cloud_options = beam_options.view_as(GoogleCloudOptions)
with beam.Pipeline(options=beam_options) as pipeline:
check_csv = (pipeline
| "ReadData" >> beam.io.ReadFromPubSub(topic=TOPIC) # Ok
| "Decode" >> beam.Map(lambda x: x.decode('utf-8')) # Ok
| "Extract informations from PubSub message" >> beam.ParDo(ReadPubSubMessage()) # Ok
.
.
.
| beam.Map(lambda x:logging.info(x))
)
我收到的错误是：
NameError: name 'external_functions' is not defined
我认为这是因为工人没有代码范围，只有任务代码。
如何在流作业数据流中编写全局函数？或者分享一个在 Dataflow 的更多任务中使用全局函数的基本示例？
谢谢你的时间
我创建了一些代码来模拟这种情况。我创建了另一个 python 文件，其中包含一个我调用的函数（在导入 lib 之后），但我遇到了同样的错误。
我也尝试过在 main 中定义函数，但显然不起作用。
下面的 main.py
import apache_beam as beam
import logging
# import library_list as external_functions
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions
# class stepGeneral(beam.DoFn):
# def process(self, element):
# variable = "External function"
# logging.info("Called general method")
# yield variable
TOPIC = "TOPIC NAME"
class step1(beam.DoFn):
def process(self, element):
variable = "first"
logging.info("This is the " + variable + " step")
yield variable
class step2(beam.DoFn):
def process(self, element):
variable = "second"
logging.info("This is the " + variable + " step")
# stepGeneral.process()
external_functions.stepGeneral()
yield variable
beam_options = PipelineOptions()
google_cloud_options = beam_options.view_as(GoogleCloudOptions)
with beam.Pipeline(options=beam_options) as pipeline:
(pipeline
| "ReadData" >> beam.io.ReadFromPubSub(topic=TOPIC) # Ok
| "First step" >> beam.ParDo(step1())
| "Second step" >> beam.ParDo(step2())
# | "window" >> beam.WindowInto(beam.window.FixedWindows(1))
| beam.Map(lambda x:logging.info(x))
)
在 library_list.py 下面
import logging
def stepGeneral():
variable = "External function"
logging.info("Called general method")
yield variable
【问题讨论】：
标签： google-cloud-dataflow worker global-scope global-functions