Google Cloud数据流和Cloud Functions错误-ModuleNotFoundError

Question

我正在从GCP中的Cloud Function触发数据流作业。

嵌入云功能的代码

import apache_beam as beam
import argparse

PROJECT = 'projectName'
BUCKET='bucketName'
SCHEMA = 'sr:INTEGER,abv:FLOAT,id:INTEGER,name:STRING,style:STRING,ounces:FLOAT'
DATAFLOW_JOB_NAME = 'jobName'

def discard_incomplete(data):
    """Filters out records that don't have an information."""
    return len(data['abv']) > 0 and len(data['id']) > 0 and len(data['name']) > 0 and len(data['style']) > 0


def convert_types(data):
    """Converts string values to their appropriate type."""
    data['abv'] = float(data['abv']) if 'abv' in data else None
    data['id'] = int(data['id']) if 'id' in data else None
    data['name'] = str(data['name']) if 'name' in data else None
    data['style'] = str(data['style']) if 'style' in data else None
    data['ounces'] = float(data['ounces']) if 'ounces' in data else None
    return data

def del_unwanted_cols(data):
    """Delete the unwanted columns"""
    del data['ibu']
    del data['brewery_id']
    return data

def execute(event, context):
    argv = [
      '--project={0}'.format(PROJECT),
      '--job_name={0}'.format(DATAFLOW_JOB_NAME),
      '--staging_location=gs://{0}/staging/'.format(BUCKET),
      '--temp_location=gs://{0}/staging/'.format(BUCKET),
      '--region=us-central1',
      '--runner=DataflowRunner'
   ]

    p = beam.Pipeline(argv=argv)
    input = 'gs://{0}/beers.csv'.format(BUCKET)

    (p | 'ReadData' >> beam.io.ReadFromText(input, skip_header_lines =1)
       | 'SplitData' >> beam.Map(lambda x: x.split(','))
       | 'FormatToDict' >> beam.Map(lambda x: {"sr": x[0], "abv": x[1], "ibu": x[2], "id": x[3], "name": x[4], "style": x[5], "brewery_id": x[6], "ounces": x[7]}) 
       | 'DeleteIncompleteData' >> beam.Filter(discard_incomplete)
       | 'ChangeDataType' >> beam.Map(convert_types)
       | 'DeleteUnwantedData' >> beam.Map(del_unwanted_cols)
       | 'WriteToBigQuery' >> beam.io.WriteToBigQuery(
           '{0}:sandeep_beer_test.beer_data'.format(PROJECT),
           schema=SCHEMA,
           write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND))
    p.run()
    

当执行Cloud函数时，数据流作业确实会被触发，但作业一直失败。当我检查作业日志时，我看到这条错误消息-ModuleNotFoundError：没有名为“google.cloud.functions”的模块

requirements.txt

apache-梁[gc p]

如果我在安装apache beam[gcp]后直接从Cloud shell运行它，则嵌入云函数中的python代码运行良好。

请分享您对如何克服丢失模块的数据流错误的意见。

谢谢

桑迪普

Answer 1

这可能是因为你正在通过 --save_main_session。