在Airflow DAG中构建Docker镜像并上传至GCP Artifact Registry的最佳实践

针对你的需求，以下是几种符合Airflow设计理念且能解决你遇到问题的最佳实践：

方案一：正确配置CloudBuildCreateBuildOperator（解决传参和工作区问题） #

你之前遇到的参数传递和工作区指定问题，可通过Operator的内置参数直接解决：

• 传递替换参数：使用substitutions参数，配合cloudbuild.yaml中的${_自定义变量名}占位符实现动态替换
• 指定工作区/源码：通过source字段配置Git仓库或云存储中的源码位置

示例代码 #

1. DAG内直接定义构建逻辑

from airflow import DAG
from airflow.providers.google.cloud.operators.cloud_build import CloudBuildCreateBuildOperator
from datetime import datetime

default_args = {
    'start_date': datetime(2024, 1, 1),
}

with DAG('build_and_push_image', default_args=default_args, schedule_interval=None) as dag:
    build_image = CloudBuildCreateBuildOperator(
        task_id='build_image',
        project_id='your-gcp-project',
        body={
            'source': {
                'gitSource': {
                    'url': '{{ dag_run.conf.get("git_repo_url") }}',  # 从DAG运行参数获取Git地址
                    'revision': '{{ dag_run.conf.get("git_repo_branch", "main") }}'
                }
            },
            'steps': [
                {
                    'name': 'gcr.io/cloud-builders/docker',
                    'args': [
                        'build', '-t', '${_ARTIFACT_REGISTRY_REPO}/${_IMAGE_NAME}:${_IMAGE_TAG}', '.'
                    ]
                },
                {
                    'name': 'gcr.io/cloud-builders/docker',
                    'args': ['push', '${_ARTIFACT_REGISTRY_REPO}/${_IMAGE_NAME}:${_IMAGE_TAG}']
                }
            ],
            'substitutions': {
                '_ARTIFACT_REGISTRY_REPO': '{{ dag_run.conf.get("artifact_repo") }}',
                '_IMAGE_NAME': '{{ dag_run.conf.get("image_name") }}',
                '_IMAGE_TAG': '{{ dag_run.conf.get("image_tag", "latest") }}'
            },
            'options': {
                'logging': 'CLOUD_LOGGING_ONLY'
            }
        },
        gcp_conn_id='google_cloud_default'
    )

2. 引用外部cloudbuild.yaml文件

如果已有现成的cloudbuild.yaml，只需在Operator中指定文件路径并传递替换参数：

build_image = CloudBuildCreateBuildOperator(
    task_id='build_image',
    project_id='your-gcp-project',
    config_file='/path/to/cloudbuild.yaml',
    source={
        'gitSource': {
            'url': '{{ dag_run.conf.get("git_repo_url") }}',
            'revision': '{{ dag_run.conf.get("git_repo_branch") }}'
        }
    },
    substitutions={
        '_ARTIFACT_REGISTRY_REPO': 'us-central1-docker.pkg.dev/your-project/repo',
        '_IMAGE_NAME': '{{ dag_run.conf.get("app_name") }}'
    },
    gcp_conn_id='google_cloud_default'
)

对应的cloudbuild.yaml片段：

steps:
- name: 'gcr.io/cloud-builders/docker'
  args: ['build', '-t', '${_ARTIFACT_REGISTRY_REPO}/${_IMAGE_NAME}:latest', '.']
- name: 'gcr.io/cloud-builders/docker'
  args: ['push', '${_ARTIFACT_REGISTRY_REPO}/${_IMAGE_NAME}:latest']

方案二：用PythonOperator调用Cloud Build API（高度自定义场景） #

如果Operator的封装不够灵活，直接调用Google Cloud Build的Python客户端可实现完全自定义的构建逻辑，比如根据配置文件动态调整步骤：

from airflow import DAG
from airflow.operators.python import PythonOperator
from google.cloud import cloudbuild_v1
from datetime import datetime

def trigger_cloud_build(**context):
    config = context['dag_run'].conf
    client = cloudbuild_v1.CloudBuildClient()
    
    build = cloudbuild_v1.Build()
    # 配置Git源码
    build.source.git_source.url = config['git_repo_url']
    build.source.git_source.revision = config.get('git_repo_branch', 'main')
    
    # 构建步骤
    build.steps = [
        {
            'name': 'gcr.io/cloud-builders/docker',
            'args': [
                'build', '-t', f"{config['artifact_repo']}/{config['image_name']}:{config.get('image_tag', 'latest')}", '.'
            ]
        },
        {
            'name': 'gcr.io/cloud-builders/docker',
            'args': ['push', f"{config['artifact_repo']}/{config['image_name']}:{config.get('image_tag', 'latest')}"]
        }
    ]
    
    # 提交构建请求
    operation = client.create_build(project_id='your-gcp-project', build=build)
    operation.result()  # 等待构建完成

default_args = {'start_date': datetime(2024, 1, 1)}
with DAG('custom_build_image', default_args=default_args, schedule_interval=None) as dag:
    custom_build = PythonOperator(
        task_id='custom_build',
        python_callable=trigger_cloud_build,
        provide_context=True,
        gcp_conn_id='google_cloud_default'
    )

方案三：使用Kaniko在K8s Pod内构建镜像（替代挂载docker.sock） #

挂载docker.sock会破坏容器隔离，直接影响主机，不符合Airflow和Kubernetes的设计原则。推荐使用Kaniko——它无需Docker daemon，可在容器内直接构建并推送镜像到Artifact Registry：

from airflow import DAG
from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from datetime import datetime

default_args = {'start_date': datetime(2024, 1, 1)}
with DAG('kaniko_build_image', default_args=default_args, schedule_interval=None) as dag:
    kaniko_build = KubernetesPodOperator(
        task_id='kaniko_build',
        image='gcr.io/kaniko-project/executor:latest',
        cmds=['/kaniko/executor'],
        arguments=[
            '--dockerfile=Dockerfile',
            '--context={{ dag_run.conf.get("git_repo_url") }}#{{ dag_run.conf.get("git_repo_branch", "main") }}',
            '--destination={{ dag_run.conf.get("artifact_repo") }}/{{ dag_run.conf.get("image_name") }}:{{ dag_run.conf.get("image_tag", "latest") }}',
            '--cache=true'  # 可选：开启构建缓存加速
        ],
        namespace='airflow',
        service_account_name='airflow-kaniko-sa',  # 需要配置有权限推送Artifact Registry的SA
        get_logs=True,
        is_delete_operator_pod=True
    )

注意：需给K8s的ServiceAccount绑定Artifact Registry的推送权限（roles/artifactregistry.writer）。

方案选择建议 #

• 若已有成熟的Cloud Build配置文件，优先选方案一，简洁易维护
• 需要高度自定义构建逻辑（比如根据配置文件动态调整步骤），选方案二
• 不想依赖GCP Cloud Build服务，希望在Airflow所在K8s集群内完成构建，选方案三

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