在GCP Cloud Composer的Airflow中调用带标签的Dataform工作流遇字段错误求助

错误根源 #

你使用的Cloud Composer 1.20.2（对应Airflow 2.2.5）依赖的apache-airflow-providers-google版本过低，该版本的Dataform Operator对应的API协议中，InvocationConfig结构体还未支持includedTags/included_Tags字段，因此触发了ValueError。

解决方法 #

方法一：升级Cloud Composer环境

将Composer升级到2.x版本（如2.0及以上），对应Airflow 2.3+及更高版本的apache-airflow-providers-google（10.0.0+），这些版本已支持Dataform工作流调用中的标签过滤参数。升级后，需将代码中的字段名修正为标准下划线格式included_tags，示例修改如下：

create_workflow_invocation = DataformCreateWorkflowInvocationOperator(
    task_id='create_workflow_invocation',
    project_id=PROJECT_ID,
    region=REGION,
    repository_id=REPOSITORY_ID,
    workflow_invocation={
        "compilation_result": "{{ task_instance.xcom_pull('create_compilation_result')['name'] }}",
        "invocation_config": { "included_tags": ["dcm_raw"], "transitive_dependencies_included": True }
    },
    timeout=None,
)

方法二：绕过Airflow Operator，用BashOperator调用gcloud命令

若无法升级Composer环境，可直接通过gcloud命令触发Dataform工作流，示例代码如下：

from airflow.operators.bash import BashOperator

create_workflow_invocation = BashOperator(
    task_id='create_workflow_invocation',
    bash_command="""
        # 获取最新的编译结果ID
        COMPILATION_RESULT=$(gcloud dataform compilation-results list \
            --project={PROJECT_ID} \
            --region={REGION} \
            --repository={REPOSITORY_ID} \
            --git-commitish={GIT_COMMITISH} \
            --limit=1 \
            --format="value(name)")
        
        # 调用Dataform工作流并指定标签
        gcloud dataform workflow-invocations create \
            --project={PROJECT_ID} \
            --region={REGION} \
            --repository={REPOSITORY_ID} \
            --compilation-result="$COMPILATION_RESULT" \
            --included-tags=dcm_raw \
            --transitive-dependencies-included
    """.format(
        PROJECT_ID=PROJECT_ID,
        REGION=REGION,
        REPOSITORY_ID=REPOSITORY_ID,
        GIT_COMMITISH=GIT_COMMITISH
    ),
    timeout=None
)

注意：需确保Composer环境的服务账号拥有dataform.workflowInvocations.create等Dataform相关权限。

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