气流2.0.1/Python 3.7.9自定义挂钩的ModuleNotFoundError

Question

在我的本地气流构建中，我的结构如下所示

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

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

从jhook导入UtilTriggers作为触发器

当我检查气流界面时，我得到了客户的坏dag，提到错误为 ModuleNotFoundError:没有名为jhook的模块

composer 1.9上也有同样的代码，目前我正在本机上运行

此外，我还尝试添加了init.py文件，并创建了一个新的文件夹job\u触发器，在该文件夹下添加的文件仍然无法工作

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

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

当一切正常时，请指导我该模块查找错误的原因

Answer 1

根据您的评论，您收到的错误消息是“import error”，似乎问题只与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.')

新1.py

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

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

回答你的问题，如果这是一个库伯内特斯问题，因为它是托管在那里。 这个问题与库伯内特斯无关

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