airflow – Page 2 – Tarik Billa

Python script scheduling in airflow

December 21, 2023 by Tarik

You should probably use the PythonOperator to call your function. If you want to define the function somewhere else, you can simply import it from a module as long as it’s accessible in your PYTHONPATH. from airflow import DAG from airflow.operators.python_operator import PythonOperator from my_script import my_python_function dag = DAG(‘tutorial’, default_args=default_args) PythonOperator(dag=dag, task_id=’my_task_powered_by_python’, provide_context=False, python_callable=my_python_function, … Read more

Airflow signals SIGTERM to subprocesses unexpectedly

December 17, 2023 by Tarik

I followed the answer here. The idea is the same: Not letting Airflow close threads too early: export AIRFLOW__CORE__KILLED_TASK_CLEANUP_TIME=604800 did the trick.

Example DAG gets stuck in “running” state indefinitely

December 10, 2023 by Tarik

To run any DAGs, you need to make sure two processes are running: airflow webserver airflow scheduler If you only have airflow webserver running, the UI will show DAGs as running, but if you click on the DAG, none of it’s tasks are actually running or scheduled, but rather in a Null state. What this … Read more

How to define Airflow DAG/task that shouldn’t run periodically

December 3, 2023 by Tarik

Based on the documentation, you can set the scheduler preset to None (Don’t schedule, use for exclusively “externally triggered” DAGs). Also, you can set it to @once if schedule once and only once.

psycopg2 could not translate host name

November 29, 2023 by Tarik

psycopg2 could not translate host name

Importing local module (python script) in Airflow DAG

November 23, 2023 by Tarik

This usually has to do with how Airflow is configured. In airflow.cfg, make sure the path in airflow_home is correctly set to the path the Airflow directory strucure is in. Then Airflow scans all subfolders and populates them so that modules can be found. Otherwise, just make sure the folder you are trying to import … Read more

Airflow not loading dags in /usr/local/airflow/dags

September 25, 2023 by Tarik

My dag is being loaded but I had the name of the DAG wrong. I was expecting the dag to be named by the file but the name is determined by the first argument to the DAG constructor dag = DAG( ‘tutorial’, default_args=default_args, schedule_interval=timedelta(1))

Airflow tasks get stuck at “queued” status and never gets running

September 23, 2023 by Tarik

Tasks getting stuck is, most likely, a bug. At the moment (<= 1.9.0alpha1) it can happen when a task cannot even start up on the (remote) worker. This happens for example in the case of an overloaded worker or missing dependencies. This patch should resolve that issue. It is worth investigating why your tasks do … Read more

Running google colab every day at a specific time

September 23, 2023 by Tarik

You need to create a notebooks.csv listing all the Colaboratory URLs. Then use colabctl to run each notebook (In order, synchronously mentioned in the CSV) and then pauses for a period of n seconds of time before running them again. You can then run python colabctl.py <end-string> <sleep-seconds>, There’s a gCookies.pkl file in the repo. … Read more

Accessing configuration parameters passed to Airflow through CLI

September 17, 2023 by Tarik

This is probably a continuation of the answer provided by devj. At airflow.cfg the following property should be set to true: dag_run_conf_overrides_params=True While defining the PythonOperator, pass the following argument provide_context=True. For example: get_row_count_operator = PythonOperator(task_id=’get_row_count’, python_callable=do_work, dag=dag, provide_context=True) Define the python callable (Note the use of **kwargs): def do_work(**kwargs): table_name = kwargs[‘dag_run’].conf.get(‘table_name’) # Rest … Read more