使用 kubernetes 执行器时找不到 Airflow-dag_id 问题

Question

我正在使用气流稳定 helm chart 并使用 Kubernetes Executor，正在为 dag 安排新的 pod，但找不到 dag_id 失败的问题。我正在使用 git-sync 来获取 dags。以下是错误和 kubernetes 配置值。有人可以帮我解决这个问题吗？

错误：

[2020-07-01 23:18:36,939] {__init__.py:51} INFO - Using executor LocalExecutor
[2020-07-01 23:18:36,940] {dagbag.py:396} INFO - Filling up the DagBag from /opt/airflow/dags/dags/etl/sampledag_dag.py
Traceback (most recent call last):
  File "/home/airflow/.local/bin/airflow", line 37, in <module>
    args.func(args)
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/cli.py", line 75, in wrapper
    return f(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/bin/cli.py", line 523, in run
    dag = get_dag(args)
  File "/home/airflow/.local/lib/python3.6/site-packages/airflow/bin/cli.py", line 149, in get_dag
    'parse.'.format(args.dag_id))
airflow.exceptions.AirflowException: dag_id could not be found: sampledag  . Either the dag did not exist or it failed to parse.

配置：

      AIRFLOW__KUBERNETES__DELETE_WORKER_PODS: false
      AIRFLOW__KUBERNETES__GIT_REPO: git@git.com/dags.git
      AIRFLOW__KUBERNETES__GIT_BRANCH: master
      AIRFLOW__KUBERNETES__GIT_DAGS_FOLDER_MOUNT_POINT: /dags
      AIRFLOW__KUBERNETES__GIT_SSH_KEY_SECRET_NAME: git-secret
      AIRFLOW__KUBERNETES__WORKER_CONTAINER_REPOSITORY: airflow-repo
      AIRFLOW__KUBERNETES__WORKER_CONTAINER_TAG: tag
      AIRFLOW__KUBERNETES__RUN_AS_USER: "50000"

采样标签

import logging
import datetime

from airflow import models
from airflow.contrib.operators import kubernetes_pod_operator
import os

args = {
    'owner': 'airflow'
}

YESTERDAY = datetime.datetime.now() - datetime.timedelta(days=1)


try:
    print("Entered try block")
    with models.DAG(
            dag_id='sampledag',
            schedule_interval=datetime.timedelta(days=1),
            start_date=YESTERDAY) as dag:
     
        print("Initialized dag")
               kubernetes_min_pod = kubernetes_pod_operator.KubernetesPodOperator(
                # The ID specified for the task.
                task_id='trigger-task',
                # Name of task you want to run, used to generate Pod ID.
                name='trigger-name',
                namespace='scheduler',
                in_cluster = True,
     
                cmds=["./docker-run.sh"],
                is_delete_operator_pod=False,
                image='imagerepo:latest',
                image_pull_policy='Always',
                dag=dag)

        print("done")

except Exception as e:
    print(str(e))
    logging.error("Error at {}, error={}".format(__file__, str(e)))
    raise

Answer 1

我有同样的问题。我通过在我的配置中添加以下内容来解决它：

AIRFLOW__KUBERNETES__DAGS_VOLUME_SUBPATH: repo/

发生的事情是 init 容器将在 [AIRFLOW__KUBERNETES__GIT_DAGS_FOLDER_MOUNT_POINT]/[AIRFLOW__KUBERNETES__GIT_SYNC_DEST] 和 AIRFLOW__KUBERNETES__GIT_SYNC_DEST 中下载你的 dag，默认情况下是 repo (https://airflow.apache.org/docs/stable/configurations-ref.html#git-sync-dest)

Answer 2

我猜测问题可能是由于您的设置不同导致：/opt/airflow/dags/dags/etl/sampledag_dag.py 和 AIRFLOW__KUBERNETES__GIT_DAGS_FOLDER_MOUNT_POINT: /dags

我会仔细检查这些是否符合您的要求，是否符合您的期望。

Answer 3

我在尝试使用稳定的 helm 气流图使用 Kubernetes Executor 时遇到了同样的问题。就我而言，我能够通过更改来解决它 AIRFLOW__KUBERNETES__RUN_AS_USER: "50000" 到 AIRFLOW__KUBERNETES__GIT_SYNC_RUN_AS_USER: "65533" 在 helm 图表的 env 部分。

link 中提到了相同的值

我得出这个结论是因为在临时工作 pod 出现之前正在运行的 init 容器 (git sync) 无法将 git dag 克隆/同步到工作 pod。就我而言，存在权限错误（即使正确传递了 ssh 克隆的 kube 密码）

注意：

• git-sync init container 即使无法获取 DAG 也不会返回错误
• Kubernetes debugging information for init containers

kubectl get pods -n [NAMESPACE]
kubectl logs -n [NAMESPACE] [POD_ID] -c git-sync

Answer 4

遇到同样的问题，我通过@gtrip 的建议解决了这个问题，将 git-sync run 用户的 UID 设置为 65533。

我会添加以下调试提示：

• git-sync init container 即使无法获取 DAG 也不会返回错误
• Kubernetes debugging information for init containers

kubectl get pods -n [NAMESPACE]
kubectl logs -n [NAMESPACE] [POD_ID] -c git-sync