在Google Composer中调用多个Cloud Functions的实现方案咨询

问题背景 #

我正在Google Composer中编写DAG，用于在工作流中调用两个Cloud Functions。基于SimpleHTTPOperator创建了如下自定义类：

class cfSFTP2GCSOp(SimpleHttpOperator):
    def execute(self, context):
        http = HttpHook(self.method, http_conn_id=self.http_conn_id)
        self.log.info("Calling HTTP method")
        target_audience = 'https://dw-etl-transactor-unzip-files-nowvpwp6oq-uc.a.run.app'
        request = google.auth.transport.requests.Request()
        idt = id_token.fetch_id_token(request, target_audience)
        self.headers = {'Authorization': "Bearer " + idt}
        response = http.run(self.endpoint,
                            self.data,
                            self.headers,
                            self.extra_options)
        self.log.info(response)

        if response == "<Response [200]>":
            return True
        else:
            return False

并通过以下任务调用该Cloud Function：

gcp_cf_dw_etl_transactor_unzip_files = cfSFTP2GCSOp(
    task_id='gcp_cf_dw_etl_unzip_files',
    method='POST',
    http_conn_id='gcp_cf_dw_etl_unzip_files',
    data={},
    endpoint='/',
    headers={},
    response_check=lambda response: True if response == "<Response [200]>" is True else False,
    dag=dag,
)

目前我通过两个任务分别调用所需的Cloud Functions，若需调用更多Cloud Functions，是否可在单个任务中完成，还是需沿用当前的多任务方式？

解决方案分析 #

1. 可以在单个任务中调用多个Cloud Functions #

你可以通过两种方式实现单任务批量调用：

方式一：修改自定义Operator，支持批量调用

扩展你的cfSFTP2GCSOp类，让它接受一个包含多个CF配置的参数，在execute方法中循环遍历配置，依次调用每个Cloud Function：

class BatchCFCallOp(SimpleHttpOperator):
    def __init__(self, cf_configs, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.cf_configs = cf_configs

    def execute(self, context):
        request = google.auth.transport.requests.Request()
        for config in self.cf_configs:
            http = HttpHook(config['method'], http_conn_id=config['http_conn_id'])
            self.log.info(f"Calling Cloud Function: {config['target_audience']}")
            idt = id_token.fetch_id_token(request, config['target_audience'])
            headers = {'Authorization': "Bearer " + idt}
            response = http.run(config['endpoint'],
                                config.get('data', {}),
                                headers,
                                config.get('extra_options'))
            self.log.info(response)
            
            # 响应检查，失败则抛出异常终止任务
            if str(response) != "<Response [200]>":
                raise Exception(f"Cloud Function call failed: {config['target_audience']}, response: {response}")
        return True

创建任务时传入多组CF配置：

batch_cf_task = BatchCFCallOp(
    task_id='batch_call_cfs',
    cf_configs=[
        {
            'method': 'POST',
            'http_conn_id': 'gcp_cf_dw_etl_unzip_files',
            'target_audience': 'https://dw-etl-transactor-unzip-files-nowvpwp6oq-uc.a.run.app',
            'endpoint': '/',
            'data': {}
        },
        {
            'method': 'POST',
            'http_conn_id': 'gcp_cf_another_function',
            'target_audience': 'https://your-another-cf-url.a.run.app',
            'endpoint': '/',
            'data': {}
        }
    ],
    dag=dag
)

方式二：使用PythonOperator编写调用逻辑

如果不想修改Operator，直接用PythonOperator编写函数批量调用：

from airflow.operators.python import PythonOperator

def call_multiple_cfs():
    request = google.auth.transport.requests.Request()
    cf_list = [
        {
            'method': 'POST',
            'http_conn_id': 'gcp_cf_dw_etl_unzip_files',
            'target_audience': 'https://dw-etl-transactor-unzip-files-nowvpwp6oq-uc.a.run.app',
            'endpoint': '/',
            'data': {}
        },
        # 添加更多CF配置
    ]
    for cf in cf_list:
        http = HttpHook(cf['method'], http_conn_id=cf['http_conn_id'])
        idt = id_token.fetch_id_token(request, cf['target_audience'])
        headers = {'Authorization': "Bearer " + idt}
        response = http.run(cf['endpoint'], cf['data'], headers)
        if str(response) != "<Response [200]>":
            raise Exception(f"CF call failed: {cf['target_audience']}")

batch_cf_task = PythonOperator(
    task_id='batch_call_cfs',
    python_callable=call_multiple_cfs,
    dag=dag
)

2. 沿用多任务方式的优势 #

单任务批量调用虽然可行，但多任务方式有不可替代的价值：

• 独立监控：每个CF调用是单独任务，在Airflow UI中可清晰查看每个任务的状态，便于排查问题。
• 并行执行：无依赖的CF可并行调用，提升工作流效率。
• 独立重试：单个CF调用失败时，仅需重试对应任务，无需重新执行所有调用。
• 清晰的工作流结构：多任务DAG更符合Airflow任务拆分的设计理念，结构直观易懂。

选择建议 #

• 如果CF之间有强依赖（必须顺序执行，一个失败全部终止），且不需要单独监控每个CF状态，可选择单任务批量调用。
• 如果需要独立监控、重试或并行执行，建议继续使用多任务方式，每个CF对应一个任务。

内容的提问来源于stack exchange，提问作者Marcelo Hirota