Upvoted both the question and the answer, but I think that this can be made a little more clear for those users who just want to pass small data objects between PythonOperator tasks in their DAGs. Referencing this question and this XCom example got me to the following solution. Super simple: from datetime import datetime … Read more
Airflow can be a bit tricky to setup. Do you have the airflow scheduler running? Do you have the airflow webserver running? Have you checked that all DAGs you want to run are set to On in the web ui? Do all the DAGs you want to run have a start date which is in … Read more
Airflow 2.x Airflow provides a branching decorator that allows you to return the task_id (or list of task_ids) that should run: @task.branch(task_id=”branch_task”) def branch_func(ti): xcom_value = int(ti.xcom_pull(task_ids=”start_task”)) if xcom_value >= 5: return “big_task” # run just this one task, skip all else elif xcom_value >= 3: return [“small_task”, “warn_task”] # run these, skip all else … Read more
Upgrade to airflow version 1.8 and use catchup_by_default=False in the airflow.cfg or apply catchup=False to each of your dags. https://github.com/apache/incubator-airflow/blob/master/UPDATING.md#catchup_by_default
The BashOperator‘s bash_command argument is a template. You can access execution_date in any template as a datetime object using the execution_date variable. In the template, you can use any jinja2 methods to manipulate it. Using the following as your BashOperator bash_command string: # pass in the first of the current month some_command.sh {{ execution_date.replace(day=1) }} … Read more
The other answers are quite technical and hard to understand. I was in your position before so I’ll explain in simple terms. Airflow can do anything. It has BashOperator and PythonOperator which means it can run any bash script or any Python script. It is a way to organize (setup complicated data pipeline DAGs), schedule, … Read more
Edit 8/27/18 – Airflow 1.10 is now released on PyPI! https://pypi.org/project/apache-airflow/1.10.0/ How to delete a DAG completely We have this feature now in Airflow ≥ 1.10! The PR #2199 (Jira: AIRFLOW-1002) adding DAG removal to Airflow has now been merged which allows fully deleting a DAG’s entries from all of the related tables. The core … Read more
When you startup airflow, make sure you set: load_examples = False inside your airflow.cfg If you have already started airflow with this not set to false, you can set it to false and run airflow resetdb in the cli (!which will destroy all current dag information!). Alternatively you can go into the airflow_db and manually … Read more
Here is how I did it with a similar request without any subdags: First create a method that returns whatever values you want def values_function(): return values Next create method that will generate the jobs dynamically: def group(number, **kwargs): #load the values if needed in the command you plan to execute dyn_value = “{{ task_instance.xcom_pull(task_ids=”push_func”) … Read more