What is Apache Airflow?
Apache Airflow is a platform to programmatically author, schedule, and monitor workflows. Use airflow to author workflows as directed acyclic graphs (DAGs) of tasks. The airflow scheduler executes your tasks on an array of workers while following the specified dependencies. Rich command line utilities make performing complex surgeries on DAGs a snap. The rich user interface makes it easy to visualize pipelines running in production, monitor progress, and troubleshoot issues when needed. When workflows are defined as code, they become more maintainable, versionable, testable, and collaborative.
More details on Apache Airflow can be found on the official website .
Nomad Job Definition
DB Backend Configuration
job "db-airflow" {
datacenters = ["global"]
type = "service"
update {
max_parallel = 1
min_healthy_time = "10s"
healthy_deadline = "5m"
progress_deadline = "30m"
auto_revert = false
canary = 0
}
group "postgres" {
count = 1
ephemeral_disk {
size = 2048
}
restart {
attempts = 5
interval = "10m"
delay = "15s"
mode = "fail"
}
network {
mode = "host"
port "postgres" {
static = 5432
// to = 5432
}
}
service {
name = "postgres-airflow"
port = "postgres"
provider = "consul"
check {
name = "Postgres TCP Check"
type = "tcp"
port = "postgres"
interval = "15s"
timeout = "2s"
}
check {
name = "Postgres CLI Check"
type = "script"
command = "/usr/bin/pg_isready"
args = ["-U", "airflow"]
interval = "10s"
timeout = "30s"
on_update = "ignore_warnings"
task = "postgres"
}
}
task "postgres" {
driver = "docker"
env {
# Postgres Specific ENV Vars
POSTGRES_USER = "airflow"
POSTGRES_PASSWORD = "airflow"
POSTGRES_DB = "airflow"
}
config {
ports = ["postgres"]
image = "postgres:13"
network_mode = "host"
volumes = [
"local/postgres-db-volume:/var/lib/postgresql/data"
]
}
resources {
cpu = 1000
memory = 2048
}
}
}
group "redis" {
count = 1
ephemeral_disk {
size = 1024
}
network {
mode = "host"
port "redis" {
static = 6379
// to = 6379
}
}
restart {
attempts = 5
interval = "10m"
delay = "15s"
mode = "fail"
}
service {
name = "redis-airflow"
port = "redis"
provider = "consul"
check {
name = "Redis TCP Check"
type = "tcp"
port = "redis"
interval = "15s"
timeout = "2s"
}
check {
name = "Redis CLI Check"
type = "script"
command = "/usr/local/bin/redis-cli"
args = ["-p", "6379", "ping"]
interval = "10s"
timeout = "30s"
on_update = "ignore_warnings"
task = "redis"
}
}
task "redis" {
driver = "docker"
config {
image = "redis:latest"
network_mode = "host"
ports = ["redis"]
args = ["--port", "6379"]
}
resources {
cpu = 1000
memory = 2048
}
}
}
}
Airflow Job Definition
job "airflow" {
datacenters = ["global"]
type = "service"
update {
max_parallel = 1
min_healthy_time = "10s"
healthy_deadline = "5m"
progress_deadline = "30m"
auto_revert = true
auto_promote = true
canary = 1
}
group "statsd-exporter" {
count = 1
network {
mode = "host"
port "afstatsdudp" {
static = 8125
}
port "afstatsdtcp" {
static = 9102
}
}
service {
name = "statsd-airflow"
port = "afstatsdudp"
provider = "consul"
}
service {
name = "statsd-airflow-scrape"
port = "afstatsdtcp"
provider = "consul"
}
restart {
attempts = 5
interval = "10m"
delay = "15s"
mode = "fail"
}
task "airflow-statsd-exporter" {
driver = "docker"
config {
image = "prom/statsd-exporter:latest"
network_mode = "host"
force_pull = true
args = ["--statsd.listen-udp=:8125", "--web.listen-address=:9102"]
}
resources {
cpu = 1000
memory = 1024
}
}
}
group "flower" {
count = 1
ephemeral_disk {
size = 1024
}
network {
mode = "host"
port "afflower" {
static = 5555
// to = 8080
}
}
service {
name = "flower-airflow"
port = "afflower"
provider = "consul"
tags = ["traefik.enable=true"]
}
restart {
attempts = 5
interval = "10m"
delay = "15s"
mode = "fail"
}
task "airflow-flower" {
driver = "docker"
env = {
# Common ENV Vars
AIRFLOW__CORE__EXECUTOR = "CeleryExecutor"
AIRFLOW__METRICS__STATSD_ON = "true"
AIRFLOW__METRICS__STATSD_PREFIX = "airflow"
AIRFLOW__WEBSERVER__EXPOSE_CONFIG = "true"
## Backwards Compatibility with Airflow < 2.3
AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION = "true"
AIRFLOW__CORE__LOAD_EXAMPLES = "false"
AIRFLOW__API__AUTH_BACKENDS = "airflow.api.auth.backend.basic_auth,airflow.api.auth.backend.session"
AIRFLOW__SCHEDULER__ENABLE_HEALTH_CHECK = "true"
AIRFLOW_UID = "0"
AIRFLOW_GID = "0"
# Airflow-Webserver Specific ENV Vars
DUMB_INIT_SETSID = "0"
}
template {
data = <<EOH
{{ range service "redis-airflow" }}
AIRFLOW__CELERY__BROKER_URL="redis://:@{{ .Address }}:{{ .Port }}/0"
{{ end }}
{{ range service "postgres-airflow" }}
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN="postgresql+psycopg2://airflow:airflow@{{ .Address }}:{{ .Port }}/airflow"
AIRFLOW__CORE__SQL_ALCHEMY_CONN="postgresql+psycopg2://airflow:airflow@{{ .Address }}:{{ .Port }}/airflow"
AIRFLOW__CELERY__RESULT_BACKEND="db+postgresql://airflow:airflow@{{ .Address }}:{{ .Port }}/airflow"
{{ end }}
{{ range service "statsd-airflow" }}
AIRFLOW__METRICS__STATSD_HOST="{{ .Address }}"
AIRFLOW__METRICS__STATSD_PORT="{{ .Port }}"
{{ end }}
EOH
destination = "secrets/config.env"
env = true
}
config {
image = "apache/airflow:2.10.2"
network_mode = "host"
force_pull = true
volumes = [
// "local/config/airflow.cfg:/opt/airflow/airflow.cfg"
]
args = ["celery", "flower"]
}
resources {
cpu = 1000
memory = 1024
}
}
}
group "webserver" {
count = 1
ephemeral_disk {
size = 1024
}
network {
mode = "host"
port "afwebserver" {
static = 8080
// to = 8080
}
}
service {
name = "ws-airflow"
port = "afwebserver"
provider = "consul"
check {
name = "Airflow Web Server HTTP Check"
type = "http"
path = "/health"
interval = "15s"
timeout = "5s"
}
tags = ["traefik.enable=true"]
}
restart {
attempts = 5
interval = "10m"
delay = "15s"
mode = "fail"
}
task "airflow-webserver" {
driver = "docker"
env = {
# Common ENV Vars
AIRFLOW__CORE__EXECUTOR = "CeleryExecutor"
AIRFLOW__METRICS__STATSD_ON = "true"
AIRFLOW__METRICS__STATSD_PREFIX = "airflow"
AIRFLOW__WEBSERVER__EXPOSE_CONFIG = "true"
## Backwards Compatibility with Airflow < 2.3
AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION = "true"
AIRFLOW__CORE__LOAD_EXAMPLES = "false"
AIRFLOW__API__AUTH_BACKENDS = "airflow.api.auth.backend.basic_auth,airflow.api.auth.backend.session"
AIRFLOW__SCHEDULER__ENABLE_HEALTH_CHECK = "true"
AIRFLOW_UID = "0"
AIRFLOW_GID = "0"
# Airflow-Webserver Specific ENV Vars
DUMB_INIT_SETSID = "0"
}
template {
data = <<EOH
{{ range service "redis-airflow" }}
AIRFLOW__CELERY__BROKER_URL="redis://:@{{ .Address }}:{{ .Port }}/0"
{{ end }}
{{ range service "postgres-airflow" }}
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN="postgresql+psycopg2://airflow:airflow@{{ .Address }}:{{ .Port }}/airflow"
AIRFLOW__CORE__SQL_ALCHEMY_CONN="postgresql+psycopg2://airflow:airflow@{{ .Address }}:{{ .Port }}/airflow"
AIRFLOW__CELERY__RESULT_BACKEND="db+postgresql://airflow:airflow@{{ .Address }}:{{ .Port }}/airflow"
{{ end }}
{{ range service "statsd-airflow" }}
AIRFLOW__METRICS__STATSD_HOST="{{ .Address }}"
AIRFLOW__METRICS__STATSD_PORT="{{ .Port }}"
{{ end }}
EOH
destination = "secrets/config.env"
env = true
}
config {
image = "apache/airflow:2.10.2"
network_mode = "host"
force_pull = true
volumes = [
// "local/config/airflow.cfg:/opt/airflow/airflow.cfg"
]
args = ["webserver"]
}
resources {
cpu = 2000
memory = 2048
}
}
}
group "scheduler" {
count = 1
ephemeral_disk {
size = 1024
}
network {
mode = "host"
}
restart {
attempts = 5
interval = "10m"
delay = "15s"
mode = "fail"
}
task "airflow-scheduler" {
driver = "docker"
env = {
# Common ENV Vars
AIRFLOW__CORE__EXECUTOR = "CeleryExecutor"
AIRFLOW__METRICS__STATSD_ON = "true"
AIRFLOW__METRICS__STATSD_PREFIX = "airflow"
AIRFLOW__WEBSERVER__EXPOSE_CONFIG = "true"
## Backwards Compatibility with Airflow < 2.3
AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION = "true"
AIRFLOW__CORE__LOAD_EXAMPLES = "false"
AIRFLOW__API__AUTH_BACKENDS = "airflow.api.auth.backend.basic_auth,airflow.api.auth.backend.session"
AIRFLOW__SCHEDULER__ENABLE_HEALTH_CHECK = "true"
AIRFLOW_UID = "0"
AIRFLOW_GID = "0"
# Airflow-Scheduler Specific ENV Vars
}
template {
data = <<EOH
{{ range service "redis-airflow" }}
AIRFLOW__CELERY__BROKER_URL="redis://:@{{ .Address }}:{{ .Port }}/0"
{{ end }}
{{ range service "postgres-airflow" }}
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN="postgresql+psycopg2://airflow:airflow@{{ .Address }}:{{ .Port }}/airflow"
AIRFLOW__CORE__SQL_ALCHEMY_CONN="postgresql+psycopg2://airflow:airflow@{{ .Address }}:{{ .Port }}/airflow"
AIRFLOW__CELERY__RESULT_BACKEND="db+postgresql://airflow:airflow@{{ .Address }}:{{ .Port }}/airflow"
{{ end }}
{{ range service "statsd-airflow" }}
AIRFLOW__METRICS__STATSD_HOST="{{ .Address }}"
AIRFLOW__METRICS__STATSD_PORT="{{ .Port }}"
{{ end }}
EOH
destination = "secrets/config.env"
env = true
}
config {
image = "apache/airflow:2.10.2"
network_mode = "host"
force_pull = true
volumes = [
// "local/config/airflow.cfg:/opt/airflow/airflow.cfg"
]
args = ["scheduler"]
}
resources {
cpu = 2000
memory = 2048
}
}
}
group "worker" {
count = 2
ephemeral_disk {
size = 1024
}
network {
mode = "host"
}
restart {
attempts = 5
interval = "10m"
delay = "15s"
mode = "fail"
}
task "airflow-worker" {
driver = "docker"
env = {
# Common ENV Vars
AIRFLOW__CORE__EXECUTOR = "CeleryExecutor"
AIRFLOW__METRICS__STATSD_ON = "true"
AIRFLOW__METRICS__STATSD_PREFIX = "airflow"
AIRFLOW__WEBSERVER__EXPOSE_CONFIG = "true"
## Backwards Compatibility with Airflow < 2.3
AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION = "true"
AIRFLOW__CORE__LOAD_EXAMPLES = "false"
AIRFLOW__API__AUTH_BACKENDS = "airflow.api.auth.backend.basic_auth,airflow.api.auth.backend.session"
AIRFLOW__SCHEDULER__ENABLE_HEALTH_CHECK = "true"
AIRFLOW_UID = "0"
AIRFLOW_GID = "0"
# Airflow-Worker Specific ENV Vars
DUMB_INIT_SETSID = "0"
}
template {
data = <<EOH
{{ range service "redis-airflow" }}
AIRFLOW__CELERY__BROKER_URL="redis://:@{{ .Address }}:{{ .Port }}/0"
{{ end }}
{{ range service "postgres-airflow" }}
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN="postgresql+psycopg2://airflow:airflow@{{ .Address }}:{{ .Port }}/airflow"
AIRFLOW__CORE__SQL_ALCHEMY_CONN="postgresql+psycopg2://airflow:airflow@{{ .Address }}:{{ .Port }}/airflow"
AIRFLOW__CELERY__RESULT_BACKEND="db+postgresql://airflow:airflow@{{ .Address }}:{{ .Port }}/airflow"
{{ end }}
{{ range service "statsd-airflow" }}
AIRFLOW__METRICS__STATSD_HOST="{{ .Address }}"
AIRFLOW__METRICS__STATSD_PORT="{{ .Port }}"
{{ end }}
EOH
destination = "secrets/config.env"
env = true
}
config {
image = "apache/airflow:2.10.2"
network_mode = "host"
force_pull = true
volumes = [
// "local/config/airflow.cfg:/opt/airflow/airflow.cfg"
]
args = ["celery", "worker"]
}
resources {
cpu = 2000
memory = 2048
}
}
}
group "triggerer" {
count = 1
ephemeral_disk {
size = 1024
}
network {
mode = "host"
}
restart {
attempts = 5
interval = "10m"
delay = "15s"
mode = "fail"
}
task "airflow-triggerer" {
driver = "docker"
env = {
# Common ENV Vars
AIRFLOW__CORE__EXECUTOR = "CeleryExecutor"
AIRFLOW__METRICS__STATSD_ON = "true"
AIRFLOW__METRICS__STATSD_PREFIX = "airflow"
AIRFLOW__WEBSERVER__EXPOSE_CONFIG = "true"
## Backwards Compatibility with Airflow < 2.3
AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION = "true"
AIRFLOW__CORE__LOAD_EXAMPLES = "false"
AIRFLOW__API__AUTH_BACKENDS = "airflow.api.auth.backend.basic_auth,airflow.api.auth.backend.session"
AIRFLOW__SCHEDULER__ENABLE_HEALTH_CHECK = "true"
AIRFLOW_UID = "0"
AIRFLOW_GID = "0"
# Airflow-Triggerer Specific ENV Vars
}
template {
data = <<EOH
{{ range service "redis-airflow" }}
AIRFLOW__CELERY__BROKER_URL="redis://:@{{ .Address }}:{{ .Port }}/0"
{{ end }}
{{ range service "postgres-airflow" }}
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN="postgresql+psycopg2://airflow:airflow@{{ .Address }}:{{ .Port }}/airflow"
AIRFLOW__CORE__SQL_ALCHEMY_CONN="postgresql+psycopg2://airflow:airflow@{{ .Address }}:{{ .Port }}/airflow"
AIRFLOW__CELERY__RESULT_BACKEND="db+postgresql://airflow:airflow@{{ .Address }}:{{ .Port }}/airflow"
{{ end }}
{{ range service "statsd-airflow" }}
AIRFLOW__METRICS__STATSD_HOST="{{ .Address }}"
AIRFLOW__METRICS__STATSD_PORT="{{ .Port }}"
{{ end }}
EOH
destination = "secrets/config.env"
env = true
}
config {
image = "apache/airflow:2.10.2"
network_mode = "host"
force_pull = true
volumes = [
// "local/config/airflow.cfg:/opt/airflow/airflow.cfg"
]
args = ["triggerer"]
}
resources {
cpu = 2000
memory = 2048
}
}
}
group "init-container" {
count = 1
ephemeral_disk {
size = 1024
}
network {
mode = "host"
}
restart {
attempts = 5
interval = "10m"
delay = "15s"
mode = "fail"
}
task "airflow-init" {
driver = "docker"
env {
# Common ENV Vars
AIRFLOW__CORE__EXECUTOR = "CeleryExecutor"
AIRFLOW__METRICS__STATSD_ON = "true"
AIRFLOW__METRICS__STATSD_PREFIX = "airflow"
AIRFLOW__WEBSERVER__EXPOSE_CONFIG = "true"
## Backwards Compatibility with Airflow < 2.3
AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION = "true"
AIRFLOW__CORE__LOAD_EXAMPLES = "false"
AIRFLOW__API__AUTH_BACKENDS = "airflow.api.auth.backend.basic_auth,airflow.api.auth.backend.session"
AIRFLOW__SCHEDULER__ENABLE_HEALTH_CHECK = "true"
AIRFLOW_UID = "0"
AIRFLOW_GID = "0"
# Airflow-Init Specific ENV Vars
_AIRFLOW_DB_MIGRATE = "true"
_AIRFLOW_WWW_USER_CREATE = "true"
_AIRFLOW_WWW_USER_USERNAME = "airflow"
_AIRFLOW_WWW_USER_PASSWORD = "airflow"
_PIP_ADDITIONAL_REQUIREMENTS = ""
}
template {
data = <<EOH
{{ range service "redis-airflow" }}
AIRFLOW__CELERY__BROKER_URL="redis://:@{{ .Address }}:{{ .Port }}/0"
{{ end }}
{{ range service "postgres-airflow" }}
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN="postgresql+psycopg2://airflow:airflow@{{ .Address }}:{{ .Port }}/airflow"
AIRFLOW__CORE__SQL_ALCHEMY_CONN="postgresql+psycopg2://airflow:airflow@{{ .Address }}:{{ .Port }}/airflow"
AIRFLOW__CELERY__RESULT_BACKEND="db+postgresql://airflow:airflow@{{ .Address }}:{{ .Port }}/airflow"
{{ end }}
{{ range service "statsd-airflow" }}
AIRFLOW__METRICS__STATSD_HOST="{{ .Address }}"
AIRFLOW__METRICS__STATSD_PORT="{{ .Port }}"
{{ end }}
EOH
destination = "secrets/config.env"
env = true
}
config {
image = "apache/airflow:2.10.2"
network_mode = "host"
force_pull = true
volumes = [
// "local/config/airflow.cfg:/opt/airflow/airflow.cfg"
]
entrypoint = [
"/bin/bash",
"-c",
"exec /entrypoint airflow db migrate; exec /entrypoint airflow version;"
]
}
resources {
cpu = 2000
memory = 2048
}
}
}
}
Understanding the Deployment
The Airflow Deployment consists of the following components:
- Postgres: The backend database for Airflow.
- Redis: The message broker for Airflow.
- StatsD-Exporter: The StatsD Exporter for Airflow.
- Flower: The Flower UI for Airflow.
- Webserver: The Webserver for Airflow.
- Scheduler: The Scheduler for Airflow.
- Worker: The Worker for Airflow.
- Triggerer: The Triggerer for Airflow.
- Init-Container: The Init Container for Airflow.
We separate the deployment into multiple groups to ensure that each component is isolated and can be scaled independently. The webserver, scheduler, worker, and triggerer groups are scaled to 1, 1, 2, and 1 respectively. The init-container group is scaled to 1 to ensure that the database is initialized and the required tables are created.
It is recommended that we separate our Backend Database and Redis away from the Airflow Nomad Job to ensure that the data is persisted and can be scaled independently.
We are relying on Consul for Service Discovery and Health Checks. We are using the template block to generate the config.env file for each task. This allows us to dynamically generate the configuration based on the services that are available in the Consul Catalog.
Conclusion
Running Apache Airflow on HashiCorp Nomad is a great way to orchestrate your workflows and tasks. The flexibility and scalability of Nomad make it an ideal platform to run Airflow. The ability to dynamically generate the configuration based on the services available in the Consul Catalog makes it easy to scale and manage the Airflow deployment.
I hope you found this post useful. Please feel free to reach out to me if you have any questions or feedback. Thank you for reading!