Introduction

For anyone new to Amazon Managed Workflows for Apache Airflow (Amazon MWAA), especially those used to managing their own Apache Airflow platform, Amazon MWAA’s configuration might appear to be a bit of a black box at first. This brief post will explore Amazon MWAA’s configuration — how to inspect it and how to modify it. We will use Airflow DAGs to review an MWAA environment’s airflow.cfg file, environment variables, and Python packages.

Amazon MWAA

Apache Airflow is a popular open-source platform designed to schedule and monitor workflows. According to Wikipedia, Airflow was created at Airbnb in 2014 to manage the company’s increasingly complex workflows. From the beginning, the project was made open source, becoming an Apache Incubator project in 2016 and a top-level Apache Software Foundation project in 2019.

With the announcement of Amazon MWAA in November 2020, AWS customers can now focus on developing workflow automation while leaving the management of Airflow to AWS. Amazon MWAA can be used as an alternative to AWS Step Functions for workflow automation on AWS.

The Amazon MWAA service is available using the AWS Management Console, as well as the Amazon MWAA API using the latest versions of the AWS SDK and AWS CLI. For more information on Amazon MWAA, read my last post, Running Spark Jobs on Amazon EMR with Apache Airflow.

Source Code

The DAGs referenced in this post are available on GitHub. Using this git clone command, download a copy of this post’s GitHub repository to your local environment.

git clone --branch main --single-branch --depth 1 --no-tags \
    https://github.com/garystafford/aws-airflow-demo.git

Accessing Configuration

Environment Variables

Environment variables are an essential part of an MWAA environment’s configuration. There are various ways to examine the environment variables. You could use Airflow’s BashOperator to simply call the command, env, or the PythonOperator to call a Python iterator function, as shown below. A sample DAG, dags/get_env_vars.py, is included in the project.

	def print_env_vars():
	keys = str(os.environ.keys().replace("', '", "'|'").split("|")
	keys.sort()
	for key in keys:
	print(key)
	get_env_vars_operator = PythonOperator(
	task_id='get_env_vars_task',
	python_callable=print_env_vars
	)

view raw get_env_vars.py hosted with ❤ by GitHub

The DAG’s PythonOperator will iterate over the MWAA environment’s environment variables and output them to the task’s log. Below is a snippet of an example task’s log.

	[2020-12-25 23:59:07,170] {{standard_task_runner.py:78}} INFO – Job 272: Subtask get_env_vars_task
	[2020-12-25 23:59:08,423] {{logging_mixin.py:112}} INFO – 'AIRFLOW_CONN_AWS_DEFAULT': 'aws://'
	[2020-12-25 23:59:08,516] {{logging_mixin.py:112}} INFO – 'AIRFLOW_CONSOLE_LOGS_ENABLED': 'false'
	[2020-12-25 23:59:08,689] {{logging_mixin.py:112}} INFO – 'AIRFLOW_CONSOLE_LOG_LEVEL': 'WARNING'
	[2020-12-25 23:59:08,777] {{logging_mixin.py:112}} INFO – 'AIRFLOW_CTX_DAG_EMAIL': 'airflow@example.com'
	[2020-12-25 23:59:08,877] {{logging_mixin.py:112}} INFO – 'AIRFLOW_CTX_DAG_ID': 'get_env_vars'
	[2020-12-25 23:59:08,970] {{logging_mixin.py:112}} INFO – 'AIRFLOW_CTX_DAG_OWNER': 'airflow'
	[2020-12-25 23:59:09,269] {{logging_mixin.py:112}} INFO – 'AIRFLOW_CTX_TASK_ID': 'get_env_vars_task'
	[2020-12-25 23:59:09,357] {{logging_mixin.py:112}} INFO – 'AIRFLOW_DAG_PROCESSING_LOGS_ENABLED': 'false'
	[2020-12-25 23:59:09,552] {{logging_mixin.py:112}} INFO – 'AIRFLOW_DAG_PROCESSING_LOG_LEVEL': 'WARNING'
	[2020-12-25 23:59:09,647] {{logging_mixin.py:112}} INFO – 'AIRFLOW_ENV_NAME': 'MyAirflowEnvironment'
	[2020-12-25 23:59:09,729] {{logging_mixin.py:112}} INFO – 'AIRFLOW_HOME': '/usr/local/airflow'
	[2020-12-25 23:59:09,827] {{logging_mixin.py:112}} INFO – 'AIRFLOW_SCHEDULER_LOGS_ENABLED': 'false'
	…
	[2020-12-25 23:59:12,915] {{logging_mixin.py:112}} INFO – 'AIRFLOW__CORE__DAG_CONCURRENCY': '10000'
	[2020-12-25 23:59:12,986] {{logging_mixin.py:112}} INFO – 'AIRFLOW__CORE__EXECUTOR': 'CeleryExecutor'
	[2020-12-25 23:59:13,136] {{logging_mixin.py:112}} INFO – 'AIRFLOW__CORE__LOAD_EXAMPLES': 'False'
	[2020-12-25 23:59:13,217] {{logging_mixin.py:112}} INFO – 'AIRFLOW__CORE__PARALLELISM': '10000'
	…
	[2020-12-25 23:59:14,531] {{logging_mixin.py:112}} INFO – 'AWS_DEFAULT_REGION': 'us-east-1'
	[2020-12-25 23:59:14,565] {{logging_mixin.py:112}} INFO – 'AWS_EXECUTION_ENV': 'AWS_ECS_FARGATE'
	[2020-12-25 23:59:14,616] {{logging_mixin.py:112}} INFO – 'AWS_REGION': 'us-east-1'
	[2020-12-25 23:59:14,647] {{logging_mixin.py:112}} INFO – 'CELERY_LOG_FILE': ''
	[2020-12-25 23:59:14,679] {{logging_mixin.py:112}} INFO – 'CELERY_LOG_LEVEL': '20'
	[2020-12-25 23:59:14,711] {{logging_mixin.py:112}} INFO – 'CELERY_LOG_REDIRECT': '1'
	[2020-12-25 23:59:14,747] {{logging_mixin.py:112}} INFO – 'CELERY_LOG_REDIRECT_LEVEL': 'WARNING'

view raw airflow_env_vars.txt hosted with ❤ by GitHub

Airflow Configuration File

According to Airflow, the airflow.cfg file contains Airflow’s configuration. You can edit it to change any of the settings. The first time you run Apache Airflow, it creates an airflow.cfg configuration file in your AIRFLOW_HOME directory and attaches the configurations to your environment as environment variables.

Amazon MWAA doesn’t expose the airflow.cfg in the Apache Airflow UI of an environment. Although you can’t access it directly, you can view the airflow.cfg file. The configuration file is located in your AIRFLOW_HOME directory, /usr/local/airflow (~/airflow by default).

There are multiple ways to examine your MWAA environment’s airflow.cfg file. You could use Airflow’s PythonOperator to call a Python function that reads the contents of the file, as shown below. The function uses the AIRFLOW_HOME environment variable to locate and read the airflow.cfg. A sample DAG, dags/get_airflow_cfg.py, is included in the project.

	def print_airflow_cfg():
	with open(f"{os.getenv('AIRFLOW_HOME')}/airflow.cfg", 'r') as airflow_cfg:
	file_contents = airflow_cfg.read()
	print(f'\n{file_contents}')
	get_airflow_cfg_operator = PythonOperator(
	task_id='get_airflow_cfg_task',
	python_callable=print_airflow_cfg
	)

view raw get_airflow_cfg.py hosted with ❤ by GitHub

The DAG’s task will read the MWAA environment’s airflow.cfg file and output it to the task’s log. Below is a snippet of an example task’s log.

	[2020-12-26 00:02:57,163] {{standard_task_runner.py:78}} INFO – Job 274: Subtask get_airflow_cfg_task
	[2020-12-26 00:02:57,583] {{logging_mixin.py:112}} INFO –
	[core]
	# The folder where your airflow pipelines live, most likely a
	# subfolder in a code repository
	# This path must be absolute
	dags_folder = /usr/local/airflow/dags
	# The folder where airflow should store its log files
	# This path must be absolute
	base_log_folder = /usr/local/airflow/logs
	# Airflow can store logs remotely in AWS S3, Google Cloud Storage or Elastic Search.
	# Set this to True if you want to enable remote logging.
	remote_logging = True
	# Users must supply an Airflow connection id that provides access to the storage
	# location.
	remote_log_conn_id = aws_default
	remote_base_log_folder = cloudwatch://arn:aws:logs:::log-group:airflow-logs:*
	encrypt_s3_logs = False
	# Logging level
	logging_level = INFO
	# Logging level for Flask-appbuilder UI
	fab_logging_level = WARN
	# Logging class
	# Specify the class that will specify the logging configuration
	# This class has to be on the python classpath
	# Example: logging_config_class = my.path.default_local_settings.LOGGING_CONFIG
	logging_config_class = log_config.LOGGING_CONFIG
	…
	# The amount of parallelism as a setting to the executor. This defines
	# the max number of task instances that should run simultaneously
	# on this airflow installation
	parallelism = 32
	# The number of task instances allowed to run concurrently by the scheduler
	dag_concurrency = 16
	…
	[aws_mwaa]
	redirect_url = https://console.aws.amazon.com/
	session_duration_minutes = 720
	[webserver]
	# The base url of your website as airflow cannot guess what domain or
	# cname you are using. This is used in automated emails that
	# airflow sends to point links to the right web server
	base_url = http://localhost:8080
	# Default timezone to display all dates in the RBAC UI, can be UTC, system, or
	# any IANA timezone string (e.g. Europe/Amsterdam). If left empty the
	# default value of core/default_timezone will be used
	# Example: default_ui_timezone = America/New_York
	default_ui_timezone = UTC
	# The ip specified when starting the web server
	web_server_host = 0.0.0.0
	# The port on which to run the web server
	web_server_port = 8080
	…

view raw airflow_cfg_log.txt hosted with ❤ by GitHub

Customizing Airflow Configurations

While AWS doesn’t expose the airflow.cfg in the Apache Airflow UI of your environment, you can change the default Apache Airflow configuration options directly within the Amazon MWAA console and continue using all other settings in airflow.cfg. The configuration options changed in the Amazon MWAA console are translated into environment variables.

To customize the Apache Airflow configuration, change the default options directly on the Amazon MWAA console. Select Edit, add or modify configuration options and values in the Airflow configuration options menu, then select Save. For example, we can change Airflow’s default timezone (core.default_ui_timezone) to America/New_York.

Once the MWAA environment is updated, which may take several minutes, view your changes by re-running the DAG,dags/get_env_vars.py. Note the new configuration item on both lines 2 and 6 of the log snippet shown below. The configuration item appears on its own (AIRFLOW__CORE_DEFAULT__UI_TIMEZONE), as well as part of the AIRFLOW_CONFIG_SECRETS dictionary environment variable.

	[2020-12-26 05:00:57,756] {{standard_task_runner.py:78}} INFO – Job 293: Subtask get_env_vars_task
	[2020-12-26 05:00:58,158] {{logging_mixin.py:112}} INFO – 'AIRFLOW_CONFIG_SECRETS': '{"AIRFLOW__CORE__DEFAULT_UI_TIMEZONE":"America/New_York"}'
	[2020-12-26 05:00:58,190] {{logging_mixin.py:112}} INFO – 'AIRFLOW_CONN_AWS_DEFAULT': 'aws://'
	…
	[2020-12-26 05:01:00,537] {{logging_mixin.py:112}} INFO – 'AIRFLOW__CORE__DAG_CONCURRENCY': '10000'
	[2020-12-26 05:01:00,578] {{logging_mixin.py:112}} INFO – 'AIRFLOW__CORE__DEFAULT_UI_TIMEZONE': 'America/New_York'
	[2020-12-26 05:01:00,630] {{logging_mixin.py:112}} INFO – 'AIRFLOW__CORE__EXECUTOR': 'CeleryExecutor'
	…

view raw airflow_env_vars_updatted.txt hosted with ❤ by GitHub

Using the MWAA API

We can also make configuration changes using the MWAA API. For example, to change the default Airflow UI timezone, call the MWAA API’s update-environment command using the AWS CLI. Include the --airflow-configuration-option parameter, passing the core.default_ui_timezone key/value pair as a JSON blob.

	aws mwaa update-environment \
	–name <your_environment_name> \
	–airflow-configuration-options """{
	\"core.default_ui_timezone\": \"America/Los_Angeles\"
	}"""

view raw mwaa_update_environment.sh hosted with ❤ by GitHub

To review an environment’s configuration, use the get-environment command in combination with jq.

	aws mwaa get-environment \
	–name <your_environment_name> | \
	jq -r '.Environment.AirflowConfigurationOptions'

view raw mwaa_get_environment.sh hosted with ❤ by GitHub

Below, we see an example of the output.

	{
	"core.default_ui_timezone": "America/Los_Angeles"
	}

view raw mwaa_configuration.json hosted with ❤ by GitHub

Python Packages

Airflow is written in Python, and workflows are created via Python scripts. Python packages are a crucial part of an MWAA environment’s configuration. According to the documentation, an ‘extra package’, is a Python subpackage that is not included in the Apache Airflow base, installed on your MWAA environment. As part of setting up an MWAA environment, you can specify the location of the requirements.txt file in the Airflow S3 bucket. Extra packages are installed using the requirements.txt file.

There are several ways to check your MWAA environment’s installed Python packages and versions. You could use Airflow’s BashOperator to call the command, python3 -m pip list. A sample DAG, dags/get_py_pkgs.py, is included in the project.

	list_python_packages_operator = BashOperator(
	task_id='list_python_packages',
	bash_command='python3 -m pip list'
	)

view raw get_py_pkgs.py hosted with ❤ by GitHub

The DAG’s task will output a list of all Python packages and package versions to the task’s log. Below is a snippet of an example task’s log.

	[2020-12-26 21:53:06,310] {{bash_operator.py:136}} INFO – Temporary script location: /tmp/airflowtmp2whgp_p8/list_python_packagesxo8slhc6
	[2020-12-26 21:53:06,350] {{bash_operator.py:146}} INFO – Running command: python3 -m pip list
	[2020-12-26 21:53:06,395] {{bash_operator.py:153}} INFO – Output:
	[2020-12-26 21:53:06,750] {{bash_operator.py:157}} INFO – Package Version
	[2020-12-26 21:53:06,786] {{bash_operator.py:157}} INFO – ———————- ———
	[2020-12-26 21:53:06,815] {{bash_operator.py:157}} INFO – alembic 1.4.2
	[2020-12-26 21:53:06,856] {{bash_operator.py:157}} INFO – amqp 2.6.1
	[2020-12-26 21:53:06,898] {{bash_operator.py:157}} INFO – apache-airflow 1.10.12
	[2020-12-26 21:53:06,929] {{bash_operator.py:157}} INFO – apispec 1.3.3
	[2020-12-26 21:53:06,960] {{bash_operator.py:157}} INFO – argcomplete 1.12.0
	[2020-12-26 21:53:07,002] {{bash_operator.py:157}} INFO – attrs 19.3.0
	[2020-12-26 21:53:07,036] {{bash_operator.py:157}} INFO – Babel 2.8.0
	[2020-12-26 21:53:07,071] {{bash_operator.py:157}} INFO – billiard 3.6.3.0
	[2020-12-26 21:53:07,960] {{bash_operator.py:157}} INFO – boto3 1.16.10
	[2020-12-26 21:53:07,993] {{bash_operator.py:157}} INFO – botocore 1.19.10
	[2020-12-26 21:53:08,028] {{bash_operator.py:157}} INFO – cached-property 1.5.1
	[2020-12-26 21:53:08,061] {{bash_operator.py:157}} INFO – cattrs 1.0.0
	[2020-12-26 21:53:08,096] {{bash_operator.py:157}} INFO – celery 4.4.7
	[2020-12-26 21:53:08,130] {{bash_operator.py:157}} INFO – certifi 2020.6.20
	…
	[2020-12-26 21:53:12,260] {{bash_operator.py:157}} INFO – pandas 1.1.0
	[2020-12-26 21:53:12,289] {{bash_operator.py:157}} INFO – pendulum 1.4.4
	[2020-12-26 21:53:12,490] {{bash_operator.py:157}} INFO – pip 9.0.3
	[2020-12-26 21:53:12,522] {{bash_operator.py:157}} INFO – prison 0.1.3
	[2020-12-26 21:53:12,550] {{bash_operator.py:157}} INFO – prometheus-client 0.8.0
	[2020-12-26 21:53:12,580] {{bash_operator.py:157}} INFO – psutil 5.7.2
	[2020-12-26 21:53:12,613] {{bash_operator.py:157}} INFO – pycparser 2.20
	[2020-12-26 21:53:12,641] {{bash_operator.py:157}} INFO – pycurl 7.43.0.5
	[2020-12-26 21:53:12,676] {{bash_operator.py:157}} INFO – Pygments 2.6.1
	[2020-12-26 21:53:12,710] {{bash_operator.py:157}} INFO – PyGreSQL 5.2.1
	[2020-12-26 21:53:12,746] {{bash_operator.py:157}} INFO – PyJWT 1.7.1
	…

view raw mwaa_python_packages.txt hosted with ❤ by GitHub

Conclusion

Understanding your Amazon MWAA environment’s airflow.cfg file, environment variables, and Python packages are all important for proper Airflow platform management. This brief post learned more about Amazon MWAA’s configuration — how to inspect it using DAGs and how to modify it through the Amazon MWAA console.