How to schedule a Python script in Airflow?

Question

I have the following dir/file structure in Windows:

Learn Airflow
        |
        project
           |_dags
               |_file_mover.py
               |_first_dag.py
            |_dockerfiles
                |_Dockerfile
            |_docker-compose.yml

In file_mover.py I have a simple script that is moving some files from LocationA to LocationB. In first_dag.py I have a script that is triggering file_mover.py. So when I do docker-compose up --build in the terminal and check the webserver localhost:8080 I do see first_dag in Airflow. So when I turn that DAG on I expected the files to be moved from LocationA to LocationB e.q. the file_mover.py to be triggered.. however, that did not happen and I dont know why.

This is file_mover.py

import os
import shutil  

location_a = r'c:\data\GG\Desktop\LocationA'
location_b = r'c:\data\GG\Desktop\LocationB'

files = os.listdir(location_a)

for f in files:
    file_path = os.path.join(location_a, f)
    shutil.move(file_path, location_b)

This is first_dag.py

    try: 
     from datetime import timedelta
     from airflow import DAG
     from airflow.operators.python_operator import PythonOperator
     from datetime import datetime
     import os
     import sys
     print('All dag modules are ok.....')

except Exception as e:
    print('Error {}'.format(e))

def first_function_execute():
     os.system('python c:\data\GG\Desktop\Python Microsoft Visual Studio\Learn Airflow\project\dags\file_mover.py')
  
with DAG (
     dag_id = 'first_dag',
     schedule_interval='@daily',
     default_args={
          'owner': 'airflow',
          'retries': 1,
          'retry_delay':  timedelta(minutes=5), 
          'start_date': datetime(2021, 1, 1),
     },
     catchup=False) as f:

     first_function_execute = PythonOperator(
          task_id='first_function_execute',
          python_callable=first_function_execute)

What I ultimately want is to schedule and monitor the file_mover.py application through the Airflow localhost, however above try-out did not seem to work...

user_14569058 · Answer

As you have properly addressed, the PythonOperator should be used to run your script. I would do it as a function which will be exported and then you can simply import it from a module as long as it's accessible in your PYTHONPATH.

import os
import shutil  

def move(location_a, location_b):
    files = os.listdir(location_a)
    for f in files:
        file_path = os.path.join(location_a, f)
        shutil.move(file_path, location_b)

Then you can schedule the DAG using the default args and the schedule interval:

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

from my_script import my_python_function

default_args = {
        'owner': 'airflow',
        'depends_on_past': False,
        'start_date': datetime.today(),
        'email': ['airflow@airflow.com'],
        'email_on_failure': False,
        'email_on_retry': False,
        'retries': 1,
        'retry_delay': timedelta(minutes=5),
}

dag = DAG('tutorial', default_args=default_args,schedule_interval="* * * * *")

PythonOperator(dag=dag,
               task_id='my_move_task',
               provide_context=False,
               python_callable=move,
               op_args=['arguments_passed_to_callable'],
               op_kwargs={'keyword_argument':'which will be passed to function'})

Then before starting Airflow, you could add the path to your scripts to the PYTHONPATH like so:

export PYTHONPATH=/path/to/my/scripts/dir/:$PYTHONPATH

More information about the Python operation and how to pass arguments to the function: https://airflow.incubator.apache.org/code.html#airflow.operators.PythonOperator

How to schedule a Python script in Airflow?

1 Answers