Cloud Function触发Dataflow作业报functions_framework模块未找到错误

报错核心是你在PipelineOptions里设置了save_main_session = True，这个参数会触发Apache Beam的序列化逻辑，把Cloud Function运行时主会话内的所有模块引用、变量全部打包上传给Dataflow worker节点。
Cloud Function的运行时依赖functions_framework是平台内置的启动模块，不在你通过requirements.txt安装的第三方包路径下，序列化过程会把这个模块的引用一并打包。Dataflow worker启动作业反序列化主会话时，在自身运行环境里找不到functions_framework模块，就会抛出你看到的ModuleNotFoundError。
独立运行Dataflow作业时不存在Cloud Function的启动环境，主会话里没有functions_framework的引用，所以不会触发这个报错。

不需要改依赖，只需要调整代码配置即可解决，步骤如下：

• 删除PipelineOptions里的save_main_session = True配置。你当前的流水线全用Beam内置的Read/Write转换逻辑，没有自定义DoFn、自定义全局变量需要传递给Dataflow worker，这个配置完全没必要开启，是导致报错的直接诱因。
• 把apache_beam相关的import语句从hello_gcs函数内部移到Python文件的最顶层，避免函数作用域的无关引用被Beam的序列化逻辑捕获。
• 如果后续流水线增加了自定义处理逻辑、必须开启save_main_session，只需在Dataflow启动参数里追加--requirements_file=requirements.txt，同时确认requirements.txt里的依赖列表和worker运行需要的完全一致即可，不要漏包。

调整后不需要修改现有requirements.txt的依赖配置，重新部署Cloud Function即可正常触发作业，修复后的最小可用代码示例：

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

def hello_gcs(event, context):
    input_file = f"gs://{event['bucket']}/{event['name']}"
    output_path = 'gs://<gcs_output_path>'
    dataflow_options = [
        '--project=<project_name>',
        '--runner=DataflowRunner',
        '--region=<region>',
        '--temp_location=gs://<temp_location>'
    ]
    options = PipelineOptions(dataflow_options)

    print(f'Event ID: {context.event_id}')
    print(f'Event type: {context.event_type}')
    print(f'Bucket: {event["bucket"]}')
    print(f'File: {event["name"]}')
    print(f'Metageneration: {event["metageneration"]}')
    print(f'Created: {event["timeCreated"]}')
    print(f'Updated: {event["updated"]}')

    with beam.Pipeline(options=options) as p:
        (
            p 
            | beam.io.ReadFromText(input_file) 
            | beam.io.WriteToText(output_path, file_name_suffix='.txt')
        )

生产环境更推荐用模板方式触发，能彻底隔离Cloud Function和Dataflow的运行环境，同时大幅降低Cloud Function的冷启动耗时、减少内存占用，支持运行时动态传入输入文件路径的实现步骤如下：

1. 单独编写模板构建脚本，通过add_value_provider_argument定义运行时可动态传入的参数

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions, GoogleCloudOptions

class FileTransferOptions(PipelineOptions):
    @classmethod
    def _add_argparse_args(cls, parser):
        # 标记为运行时参数，模板创建时不需要传值，启动作业时再传入
        parser.add_value_provider_argument('--input_path', type=str)
        parser.add_value_provider_argument('--output_path', type=str)

if __name__ == '__main__':
    options = FileTransferOptions()
    gcp_options = options.view_as(GoogleCloudOptions)
    # 以下参数为模板固定配置，创建模板时指定
    gcp_options.project = '<your_project_id>'
    gcp_options.region = '<your_region>'
    gcp_options.temp_location = 'gs://<your_temp_bucket>/temp/'
    # 模板文件存储路径
    gcp_options.template_location = 'gs://<your_template_bucket>/templates/gcs-transfer.json'
    options.view_as(beam.options.pipeline_options.StandardOptions).runner = 'DataflowRunner'

    with beam.Pipeline(options=options) as p:
        (
            p
            # 直接引用运行时参数，不需要硬编码路径
            | 'ReadFile' >> beam.io.ReadFromText(options.input_path)
            | 'WriteFile' >> beam.io.WriteToText(options.output_path, file_name_suffix='.txt')
        )

2. 在安装了对应版本apache-beam[gcp]的环境（比如Cloud Shell、本地开发机）执行上述脚本，脚本会自动完成模板构建，将模板文件上传到你指定的GCS模板路径。
3. 改写Cloud Function逻辑，不再本地构建Beam流水线，而是通过Dataflow API调用已上传的模板，传入GCS事件触发的文件路径作为运行参数即可：

from google.cloud import dataflow_v1beta3
import re

def hello_gcs(event, context):
    print(f"Triggered by file: gs://{event['bucket']}/{event['name']}")
    # 构造动态路径
    input_path = f"gs://{event['bucket']}/{event['name']}"
    output_path = f"gs://<your_output_bucket>/output/{event['name']}"
    # 生成符合Dataflow命名规范的作业名（仅支持小写字母、数字、横杠）
    job_name = "gcs-transfer-" + re.sub(r'[^a-z0-9-]', '-', event['name'].lower())

    client = dataflow_v1beta3.TemplatesServiceClient()
    request = dataflow_v1beta3.LaunchTemplateRequest(
        project_id="<your_project_id>",
        location="<your_region>",
        gcs_path="gs://<your_template_bucket>/templates/gcs-transfer.json",
        launch_parameters=dataflow_v1beta3.LaunchTemplateParameters(
            job_name=job_name,
            parameters={
                "input_path": input_path,
                "output_path": output_path
            }
        )
    )
    resp = client.launch_template(request=request)
    print(f"Launched Dataflow job ID: {resp.job.id}")

4. 这个方案下Cloud Function的requirements.txt只需要保留functions-framework==3.*和google-cloud-dataflow-client即可，不需要安装整个apache-beam包，部署包体积能缩小90%以上，冷启动速度从几十秒降到几秒级。

内容的提问来源于stack exchange，提问作者Aaron Gonzalez