在Google Composer/Airflow中使用KubernetesPodOperator挂载DAG文件夹咨询

最简单实现方案：复用Google Composer内置DAG持久卷 #

不需要额外创建存储资源，Composer已经默认做好了GCS DAG存储桶到集群PVC的自动同步，直接挂载这个内置PVC即可获取最新的DAG目录文件。

修改后的代码示例如下：

from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import KubernetesPodOperator
from airflow.providers.cncf.kubernetes.volume import Volume
from airflow.providers.cncf.kubernetes.volume_mount import VolumeMount

kubernetes_min_pod = KubernetesPodOperator(
    task_id='pod-ex-minimum',
    cmds=["bash", "-c"],
    # 挂载后直接运行DAG目录下的对应Python脚本即可，这里以dags/my_script.py为例
    arguments=["python /opt/dags/my_script.py"],
    namespace='default',
    image='toru2220/scrapy-chrome:latest',
    is_delete_operator_pod=True,
    get_logs=True,
    in_cluster=False,
    volumes=[
        # 直接使用Composer环境默认自带的PVC，无需自行创建
        Volume("dag-volume", {"persistentVolumeClaim": {"claimName": "composer-user-workloads"}})
    ],
    volume_mounts=[
        VolumeMount(
            "dag-volume", 
            "/opt/dags", # 容器内的挂载路径，可根据需求自定义
            sub_path="dags", # 直接挂载PVC下的DAG根目录，也可指定到具体子目录
            read_only=True # 建议设为只读，避免误修改DAG源文件
        )
    ],
)

注意事项 #

• 上述PVC名称适用于Google Composer 2及以上版本，如果你使用的是Composer 1，将PVC名称改为airflow-dags、sub_path改为/dags即可
• 如果只需要挂载DAG下的特定子文件夹，修改sub_path参数即可，比如脚本存放在dags/project/spider路径下，就填sub_path="dags/project/spider"
• 不要在挂载的DAG目录内写入临时文件，改动会同步回GCS的DAG存储桶，影响所有DAG的正常运行

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