Apache Beam 无法识别脚本 GCP 中的全局函数

Question

我正在开展一个在 GCP 上创建流处理预测引擎的项目。我主要从这个 repo here 中学习。但是，当我尝试执行脚本 blogposts/got_sentiment/4_streaming_pipeline/streaming_tweet.py 时，我不断收到错误

NameError: name 'estimate' is not defined [while running 'generatedPtransform-129']

我的函数如下所示

from __future__ import absolute_import

import argparse
import datetime
import json
import logging

import numpy as np

import apache_beam as beam
import apache_beam.transforms.window as window
from apache_beam.io.gcp.bigquery import parse_table_schema_from_json
from apache_beam.options.pipeline_options import StandardOptions, GoogleCloudOptions, SetupOptions, PipelineOptions
from apache_beam.transforms.util import BatchElements

from googleapiclient import discovery

def init():
    ........

def estimate_cmle():

   init()
   .....

def estimate(instances):

   estimate_cmle()
   ......


def run(argv=None):
....

   output = (lines
                     | 'assign window key' >> beam.WindowInto(window.FixedWindows(10))
                     | 'batch into n batches' >> BatchElements(min_batch_size=49, max_batch_size=50)
                     | 'predict sentiment' >> beam.FlatMap(lambda messages: estimate(messages))
                     )

.....


f __name__ == '__main__':
    logging.getLogger().setLevel(logging.INFO)
    run()

这是梁似乎无法识别估计函数的地方，尽管我在同一个脚本中创建它。

编辑

尝试使用beam.FlatMap(estimate) 出现错误

name 'estimate_cmle' is not defined [while running 'generatedPtransform-1208']

Answer 1

看看这两部分：

函数定义

def estimate(instances):
......

函数调用

beam.FlatMap(lambda messages: estimate(messages,estimate_cmle))

您的调用需要一个带有 2 个参数的函数，您声明的函数只有一个。您的 python 脚本只包含一个带有 1 个参数的估计值，并且没有定义带有 2 个参数的函数。

在 repo 中的示例中，调用仅包含 1 个参数，因此可以正常工作。解决这个问题，它应该可以工作了