How to install dependency modules for airflow DAG task(or python code)? , "Failed to import module" in airflow DAG when using kuberentesExecutor

user_8089726 0
Viewed 1497

I have a airflow DAG "example_ml.py" which has a task "train_ml_model" and this task is calling/running a python script "training.py".

-Dags/example_ml.py -Dags/training.py

When I run Dag, it is failing to import modules required for training script to execute. Error in import sklearn module

Code snippet for DAG task:

   train_model = PythonOperator(
        task_id='train_model',
        python_callable=training,
        dag = dag
    )

PS: I'm using k8s cluster. Airflow is running in k8s cluster, and executor is set to kubernetesExecutor. So when each DAG is triggered a new pod gets assigned to complete the task.

kubernetespodoperator airflow
3 Answers

Could you give more details? Are you running this on your local computer? A container? Are you sure the package is installed? As you commented out, the error seems to be related to missing package. Creating a task to install may not solve the issue. The ideal is just install requirements on whatever you are running airflow

user_123012100

I had the same issue and this is how i solved it:

running following python code

>>> import sys
>>> from pprint import pprint
>>> pprint(sys.path)

I get these paths

 '/home/.local/lib/python3.6/site-packages',
 '/usr/local/lib/python3.6/dist-packages',
 '/usr/lib/python3/dist-packages'

for me airflow is listed under

'/usr/local/lib/python3.6/dist-packages'

therefore for the package to be found, it must be installed right here. I used this command to install my package:

sudo python3 -m pip install -system [package-name] -t $(pwd)
user_152933540

Best practices (to my little knowledge) is to build and provide your own custom docker image that has all the required dependencies. Once you got that you push it to a repository of your choice and then there is a specific set of option you can use in your DAG file to declare what docker image to use for each task in your DAG, as in :

def use_airflow_binary():
rc = os.system("airflow -h")
assert rc == 0
# You don't have to use any special KubernetesExecutor configuration if you don't want to
start_task = PythonOperator(
task_id="start_task", python_callable=print_stuff, dag=dag
)
# But you can if you want to
one_task = PythonOperator(
task_id="one_task", python_callable=print_stuff, dag=dag,
executor_config={"KubernetesExecutor": {"image": "airflow:latest"}}
)
# Use the airflow -h binary
two_task = PythonOperator(
task_id="two_task", python_callable=use_airflow_binary, dag=dag,
executor_config={"KubernetesExecutor": {"image": "airflow:latest"}}
)

The interesting bit here is :

executor_config={"KubernetesExecutor": {"image": "airflow:latest"}}

This is where you can use your custom built docker image.

There is even more option available (resources allocation, affinity etc ...)

All credits goes to Marc Lamberti

user_35276150
Related