进程名称错误：未定义名称“monitoring_v3”[运行“pubsub未读消息-ptransform-32”时]

Question

不能运行流数据流和获得未读消息的计数，我需要流数据流来监视未读消息，我的代码有任何问题。

import logging
import datetime
import argparse
import yaml
import apache_beam as beam
from apache_beam import Pipeline, ParDo, DoFn
from apache_beam.options.pipeline_options import PipelineOptions
from google.cloud import monitoring_v3
from google.cloud.monitoring_v3 import query

class unread(DoFn):
    def __init__(self, project, pubsub_subscription):
        self.project = project
        self.pubsub_subscription = pubsub_subscription

    def process(self, key_value):
        logging.info("Key_value {}".format(key_value))
        client = monitoring_v3.MetricServiceClient()
        result = query.Query(
            client,
            project,
            'pubsub.googleapis.com/subscription/num_undelivered_messages',
            end_time=datetime.datetime.now(),
            minutes=2,
            ).as_dataframe()
        if result.empty:
            logging.info("All Messages Read")
        else:
            logging.info(result['pubsub_subscription'][self.project][self.pubsub_subscription][0])
        return result

def run():
    pipeline_args = [arguments passed
                    ]
    logging.info("Pipeline Arguments: {}".format(pipeline_args))

    try:
        # Set `save_main_session` to True so DoFns can access globally imported modules.
        pipeline_options = PipelineOptions(
            pipeline_args, streaming=True, save_main_session=True
        )

        with Pipeline(options=pipeline_options) as pipeline:
            (
            pipeline
            | 'Initializing ' >> beam.Create(['1'])
            | 'PubSub Unread Messages' >> ParDo(unread(project, pubsub_subscription))
            )
    except Exception as e:
        logging.error("Failed to Deploy DataFlow Pipeline. Error {}".format(e))


if __name__ == "__main__":

    with open("input.yaml", 'r') as yamlfile:
        cfg = yaml.load(yamlfile, Loader=yaml.BaseLoader)

    logging.basicConfig(format='%(asctime)s %(levelname)s - %(message)s', datefmt='%y-%m-%d %H:%M:%S', level=logging.INFO)

    parser = argparse.ArgumentParser()
    known_args, pipeline_args = parser.parse_known_args()

    run()

Answer 1

在这里查看如何处理名称错误。

尝试在函数中导入模块，或者启用--save_main_session管道选项。