在Airflow DAG中导入本地模块（Python脚本）

Question

我正在尝试将本地模块（一个Python脚本）导入我的DAG中。

目录结构：

airflow/
├── dag
│   ├── __init__.py
│   └── my_DAG.py
└── script
    └── subfolder
        ├── __init__.py
        └── local_module.py

my_DAG.py文件中的示例代码：

#trying to import from local module
from script.subfolder import local_module  

#calling a function in local_module.py  
a = some_function()  

在Airflow中我遇到了一个错误，它说“Broken DAG: my_DAG. No module named 'local_module'。

我已经将Airflow更新到1.9.0版本，但这并没有解决问题。

• 这里的解决方案是什么？
• 我还在某处读到，我可以通过创建插件来解决这个问题。有人能告诉我如何做吗？

谢谢。

- hotchocolate

Answer 1

这通常涉及到Airflow的配置。

在airflow.cfg中，请确保airflow_home路径正确设置为Airflow目录结构所在的路径。

然后，Airflow会扫描所有子文件夹并填充它们，以便可以找到模块。

否则，请确保您尝试导入的文件夹在Python路径中：如何使用PYTHONPATH。

- tobi6

---

自从这篇答案发布以来，我的 airflow.cfg 文件没有 airflow_home 这一项。 - LondonRob

您还可以通过设置export AIRFLOW_HOME=path_to_my_airflow_home_dir来实现此目的。 - J.J.

非常感谢提供文档链接，特别是因为随着版本的更新，这些答案可能会过时。 - Joey Baruch

根据链接 https://airflow.apache.org/docs/apache-airflow/stable/modules_management.html#additional-modules-in-airflow，Airflow仅将“dag”、“plugins”和“config”添加到“sys.path”中，“airflow_home”不在路径中。 - shlomiLan

Answer 2

我做的方式如下：

1. 在您的子文件夹中创建一个具有main()函数的Python脚本。
2. 在您的dag文件中，包括子文件夹和文件的路径声明。

现在您可以在PythonOperator中使用此脚本。

import sys
sys.path.insert(0,"/root/airflow/dags/subfolder"))
import subfolder.script_name as script
...    
t1=PythonOperator(
    task_id='python_script',
    python_callable=script.main,
    dag=dag
)

- Shaby

---

假设文件位于dags/script/file1.py，我该如何导入它？ - Ashish Kumar

Answer 3

如果您在Docker中运行Airlow，则需要按以下方式进行操作：

1. 在dags文件夹中创建您的模块文件夹。例如：programs
2. 按照以下方式使用它（这是Docker的正确路径）：

import sys
sys.path.append('/opt/airflow/dags/programs/my_module')
import my_module
task1 = PythonOperator(
        task_id='my_task_name',
        python_callable=my_module.my_func,
        dag=dag,
    )

- RafaelJan

---

我测试了以下代码并进行了一些更改，现在可以正常运行: 我将 " sys.path.append('/opt/airflow/dags/programs/my_module') " 更改为 " sys.path.append('/opt/airflow/dags/programs/') " 看起来不需要在路径末尾添加Python文件名。 - Ahmad Karrabi

虽然不太好看，但是能用。我使用的是 sys.path.append(os.path.dirname(os.path.abspath(__file__))) 而不是路径。 - KIC