自定义挂钩的 Airflow 2.0.1/Python 3.7.9 ModuleNotFoundError

Question

我的原生气流构建中的结构如下

dags/cust_dag.py dags/jhook.py --包含类UtilTriggers，其下有多个方法

在 cust_dag 代码中，我将该钩子/模块称为：

从 jhook 导入 UtilTriggers 作为触发器

当我检查 Airflow UI 时，我因为 cust_dag 提到错误而被打破 dag ModuleNotFoundError: 没有名为 jhook 的模块

同样的代码在 composer 1.9 上运行，目前我在原生气流上运行。

我还尝试添加 init.py 文件，并创建了一个新文件夹 job_trigger，我在该文件夹下添加了该文件仍然无法正常工作。

我已经尝试过这个问题Apache Airflow DAG cannot import local module中提到的解决方案

即在钩子自定义模块和 dag 文件中添加以下代码行 导入系统 sys.path.insert(0,os.path.abspath(os.path.dirname(file)))

当一切正常时，请指导我导致此 ModuleNotFound 错误的原因。

Answer 1

根据您的 cmets，您收到的错误消息是“导入错误”，看来问题仅与 python 有关。

dag1.py

from __future__ import print_function

import datetime

from airflow import models
from airflow.operators import bash_operator
from airflow.operators import python_operator

from new1 import hi as h1

default_dag_args = {
  # The start_date describes when a DAG is valid / can be run. Set this to a
  # fixed point in time rather than dynamically, since it is evaluated every
  # time a DAG is parsed. See:
  # https://airflow.apache.org/faq.html#what-s-the-deal-with-start-date
  'start_date': datetime.datetime(2018, 1, 1),
}

# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
      'demo_run',
      schedule_interval=datetime.timedelta(days=1),
      default_args=default_dag_args) as dag:

  # An instance of an operator is called a task. In this case, the
  # hello_python task calls the "greeting" Python function.
  hello_python = python_operator.PythonOperator(
      task_id='hello_world',
      python_callable=h1.funt,
      op_kwargs={"x" : "python"})

  # Likewise, the goodbye_bash task calls a Bash script.
  goodbye_bash = bash_operator.BashOperator(
      task_id='bye',
      bash_command='echo Goodbye.')

new1.py

class hi:
   @staticmethod
   def funt(x):
       return x + " is a programming language"

1. 由于您将所有方法都用作静态方法，因此无需将 self 传递给您的方法。start 方法中的 self 关键字指的是对象。因为可以在不创建对象的情况下调用静态方法，所以它们没有 self 关键字。
2. 如果您在方法中传递一些参数，请确保参数也传递给 DAG 任务，方法是提供 op_args and op_kwargs arguments。

回答您的问题，这是否是 Kubernetes 问题，因为它托管在那里。 此问题与 Kubernetes 无关

• 当我们创建 Composer 环境时，Composer 服务会为每个环境创建一个 GKE 集群。集群是自动命名和标记的，不应由用户手动删除。集群是通过 Deployment Manager 创建和管理的。
• 如果集群被删除，那么环境将无法修复，需要重新创建。 Kubernetes 错误会像 “Http 错误状态码：400 Http 错误消息：BAD REQUEST”等