代码之家 › 专栏 › 技术社区 › Meghdeep Ray

从与执行文件位于同一指定路径的文件导入时出现问题

airflow python

Meghdeep Ray · 技术社区 · 6 年前

为了有效地构造代码,我使用 sys.path += [ PATH_TO_MODULE ] 将函数导入我的DAG。

结构如下:

- MODULE
 |_run.py
 |_aux_functions.py
 |_config.py

它在导入时抛出错误 config 在里面 aux_functions .

辅助功能 是进口的吗 run.py .

DAG代码:

from airflow.operators.python_operator import PythonOperator
from airflow                           import DAG

from MODULE_PATHS import PATH_TO_MODULE
import datetime
import sys

sys.path += [ PATH_TO_MODULE ]

from run import run_function

default_args = { "start_date": datetime.datetime( 2018, 8, 20 ) }

with DAG( "run_dag_v1", default_args= default_args, schedule_interval= "0 0 * * *", ) as dag:
    task = PythonOperator( task_id= "run_function", python_callable= run_function, provide_context= True )

task

这个 跑py 进口 辅助功能 哪些进口 settings from config 在它运行期间,这是不起作用的。出现的错误是: ImportError: cannot import name 'settings' .

# run.py
import aux_functions
.....
def run_function( **kwargs ):
    .....

# aux_functions.py
from config import settings
.....

# config.py
settings = { ..... }

当代码仅在shell中执行时,它可以正常工作,没有问题,但是当它在气流中运行时,它会不断显示此导入错误。

测试通过以下方式完成: airflow test run_dag_v1 run_function 2018-8-21

如果我移动 settings 从 配置 到 辅助功能 它工作正常,但为什么会发生这种情况,我如何避免它?

1 回复 | 直到 6 年前

Meghdeep Ray 6 年前

目录结构是这样的:

|_MODULE_1
|   |_ run.py
|   |_ aux_functions.py
|   |_ config.py
|   
|_MODULE_2
    |_ code.py
    |_ config.py

发生的事情是一个DAG正在进口 sys.path += [ PATH_MODULE_1 ] 另一个DAG正在进口 sys.path += [ PATH_MODULE_2 ] .

在气流运行期间,它的配置路径可能错误,因此两个DAG中的任何一个总是出现故障。因此出现导入错误,因为另一个配置没有 settings 对象它们从未同时正常运行。

解决方案只是确保在每种情况下 import config 是相对的: import .config .通过这样做,气流现在可以识别正确的 config.py 在每种情况下。

我发现最好的管理方法是导入父文件夹路径 from MODULE_PATHS import PATH_MODULES_DIRECTORY 加上 __init__.py 调用同一模块内的其他文件时,只使用相对导入。

所以新的DAG都有相同的 sys.path 所有模块都可以用它们的模块名导入。

MODULES_DIRECTORY
    |
    |_MODULE_1
    |   |_ __init__.py
    |   |_ run.py
    |   |_ aux_functions.py
    |   |_ config.py
    |   
    |_MODULE_2
        |_ __init__.py
        |_ code.py
        |_ config.py

from airflow.operators.python_operator import PythonOperator
from airflow                           import DAG

from MODULE_PATHS import PATH_MODULES_DIRECTORY
import datetime
import sys

sys.path += [ PATH_MODULES_DIRECTORY ]

from MODULE_1.run import run_function

default_args = { "start_date": datetime.datetime( 2018, 8, 20 ) }

with DAG( "run_dag_v1", default_args= default_args, schedule_interval= "0 0 * * *", ) as dag:
    task = PythonOperator( task_id= "run_function", python_callable= run_function, provide_context= True )

task

# run.py
import .aux_functions
.....
def run_function( **kwargs ):
    .....

# aux_functions.py
from .config import settings
.....

# config.py
settings = { ..... }