parallelism: not a very descriptive name. The description says it sets the maximum task instances for the airflow installation, which is a bit ambiguous — if I have two hosts running airflow workers, I’d have airflow installed on two hosts, so that should be two installations, but based on context ‘per installation’ here means ‘per … Read more
Here is how we manage it for our team. First in terms of naming convention, each of our DAG file name matches the DAG Id from the content of the DAG itself (including the DAG version). This is useful because ultimately it’s the DAG Id that you see in the Airflow UI so you will … Read more
In your airflow.cfg, you’ve these two configurations to control this behavior: # after how much time a new DAGs should be picked up from the filesystem min_file_process_interval = 0 dag_dir_list_interval = 60 You might have to reload the web-server, scheduler and workers for your new configuration to take effect.
You can import the logging module into your code and write to logs that way import logging logging.info(‘Hello’) Here are some more options import logging logging.debug(‘This is a debug message’) logging.info(‘This is an info message’) logging.warning(‘This is a warning message’) logging.error(‘This is an error message’) logging.critical(‘This is a critical message’)
Try the following: export AIRFLOW_GPL_UNIDECODE=yes OR export SLUGIFY_USES_TEXT_UNIDECODE=yes Using export makes the environment variable available to all the subprocesses. Also, make sure you are using pip install apache-airflow[postgres] and not pip install airflow[postgres] Which should you use: if using AIRFLOW_GPL_UNIDECODE, airflow will install a dependency that is under GPL license, which means you won’t be … Read more
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
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
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