Kubernetes executor for dynamic scaling in Apache Airflow - Mini Project: Build & Apply

Kubernetes Executor for Dynamic Scaling in Airflow

📖 Scenario: You are setting up Apache Airflow to run tasks using the Kubernetes executor. This setup allows Airflow to create new pods dynamically for each task, helping your system scale automatically based on workload.Imagine you manage a bakery that gets many orders. Instead of baking all cakes in one oven, you want to start new ovens (pods) only when needed. This project will help you configure Airflow to do just that with Kubernetes.

🎯 Goal: Build a simple Airflow configuration that uses the Kubernetes executor to run tasks dynamically on Kubernetes pods.You will create the initial Airflow configuration, add Kubernetes executor settings, write a simple DAG that runs a task, and finally run the DAG to see dynamic pod creation.

📋 What You'll Learn

Create an Airflow configuration dictionary with basic settings

Add Kubernetes executor specific configuration variables

Write a simple Airflow DAG with one task using the Kubernetes executor

Print the DAG's task status to confirm dynamic pod execution

💡 Why This Matters

🌍 Real World

Using the Kubernetes executor in Airflow helps teams run many tasks in parallel by creating pods dynamically. This is like opening new ovens only when needed in a bakery, saving resources and time.

💼 Career

Many DevOps and data engineering roles require knowledge of Airflow and Kubernetes. Understanding how to configure Airflow with Kubernetes executor is a valuable skill for managing scalable workflows.

1

Create basic Airflow configuration dictionary

Create a dictionary called airflow_config with these exact entries: 'executor': 'SequentialExecutor', 'dags_folder': '/usr/local/airflow/dags', and 'sql_alchemy_conn': 'sqlite:////usr/local/airflow/airflow.db'.

Apache Airflow

# Create the airflow_config dictionary with the specified keys and values
# Your code here

Hint

Use curly braces {} to create the dictionary and include the exact keys and values as strings.

2

Add Kubernetes executor configuration

Add these two entries to the existing airflow_config dictionary: 'executor': 'KubernetesExecutor' to replace the old executor, and 'kube_namespace': 'airflow' to specify the Kubernetes namespace.

Apache Airflow

airflow_config = {
    'executor': 'SequentialExecutor',
    'dags_folder': '/usr/local/airflow/dags',
    'sql_alchemy_conn': 'sqlite:////usr/local/airflow/airflow.db'
}
# Update airflow_config to use KubernetesExecutor and add kube_namespace
# Your code here

Hint

Replace the 'executor' value and add the 'kube_namespace' key with value 'airflow'.

3

Write a simple Airflow DAG using Kubernetes executor

Write a Python code snippet that imports DAG and PythonOperator from airflow, creates a DAG named example_k8s_dag with default args including start_date as datetime(2024, 1, 1), and adds one PythonOperator task called print_hello that runs a function printing 'Hello from Kubernetes Executor'.

Apache Airflow

airflow_config = {
    'executor': 'KubernetesExecutor',
    'dags_folder': '/usr/local/airflow/dags',
    'sql_alchemy_conn': 'sqlite:////usr/local/airflow/airflow.db',
    'kube_namespace': 'airflow'
}

# Import DAG and PythonOperator, define default_args, create DAG and task
# Your code here

Hint

Use PythonOperator with python_callable=print_hello and assign it to the DAG.

4

Print the task status to confirm dynamic pod execution

Write a print statement that outputs the string 'Task print_hello is ready to run on Kubernetes Executor'.

Apache Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def print_hello():
    print('Hello from Kubernetes Executor')

default_args = {
    'start_date': datetime(2024, 1, 1)
}

example_k8s_dag = DAG(
    'example_k8s_dag',
    default_args=default_args,
    schedule_interval='@daily'
)

print_hello = PythonOperator(
    task_id='print_hello',
    python_callable=print_hello,
    dag=example_k8s_dag
)

# Print the task status below
# Your code here

Hint

Use print() with the exact string to confirm the task setup.

Practice

1. What is the main benefit of using the Kubernetes executor in Airflow?

easy

A. It disables task retries to save resources.

B. It stores all task logs in a local file system.

C. It schedules tasks only on a single fixed server.

D. It runs each task in its own pod, allowing dynamic scaling.

Solution

1. Step 1: Understand Kubernetes executor behavior

   The Kubernetes executor runs each Airflow task in a separate pod, which allows tasks to run independently and scale dynamically.

2. Step 2: Compare options with executor features

   Options B, C, and D describe behaviors not related to Kubernetes executor's main feature of dynamic pod creation and scaling.

3. Final Answer:

   It runs each task in its own pod, allowing dynamic scaling. -> Option D

4. Quick Check:

   Kubernetes executor = dynamic pod scaling [OK]

Hint: Kubernetes executor = one pod per task for scaling [OK]

Common Mistakes:

• Thinking tasks run on a single server
• Confusing executor with local logging
• Assuming retries are disabled

2. Which configuration key is required to enable the Kubernetes executor in Airflow's airflow.cfg?

easy

A. executor = LocalExecutor

B. executor = CeleryExecutor

C. executor = KubernetesExecutor

D. executor = SequentialExecutor

Solution

1. Step 1: Identify executor configuration key

   In airflow.cfg, the executor key sets which executor Airflow uses.

2. Step 2: Match correct executor name for Kubernetes

   The correct value to enable Kubernetes executor is KubernetesExecutor. Other options are different executors.

3. Final Answer:

   executor = KubernetesExecutor -> Option C

4. Quick Check:

   Enable Kubernetes executor = executor = KubernetesExecutor [OK]

Hint: Set executor = KubernetesExecutor in airflow.cfg [OK]

Common Mistakes:

• Using LocalExecutor or CeleryExecutor by mistake
• Misspelling KubernetesExecutor
• Changing unrelated config keys

3. Given this Airflow task using Kubernetes executor, what happens when the DAG runs?

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime
with DAG('example', start_date=datetime(2024,1,1)) as dag:
    task = BashOperator(task_id='print_date', bash_command='date')

medium

A. The task runs inside a new Kubernetes pod created dynamically.

B. The task runs on the Airflow scheduler machine only.

C. The task fails because BashOperator is not supported with Kubernetes executor.

D. The task runs on a fixed worker node without pod creation.

Solution

1. Step 1: Understand task execution with Kubernetes executor

   When using Kubernetes executor, each task runs inside its own pod created dynamically in the Kubernetes cluster.

2. Step 2: Analyze the given BashOperator task

   The BashOperator runs a simple shell command. With Kubernetes executor, this command runs inside a new pod, not on the scheduler or fixed node.

3. Final Answer:

   The task runs inside a new Kubernetes pod created dynamically. -> Option A

4. Quick Check:

   BashOperator + Kubernetes executor = dynamic pod run [OK]

Hint: Kubernetes executor runs each task in a new pod [OK]

Common Mistakes:

• Assuming tasks run on scheduler machine
• Thinking BashOperator is unsupported
• Believing pods are not created dynamically

4. You configured Airflow with Kubernetes executor but tasks are not running and pods are not created. What is the most likely cause?

medium

A. The Kubernetes cluster credentials are missing or incorrect.

B. The DAG file has syntax errors unrelated to executor.

C. The Airflow scheduler is running with LocalExecutor.

D. The BashOperator command is invalid.

Solution

1. Step 1: Check Kubernetes executor requirements

   Kubernetes executor needs valid cluster credentials to create pods dynamically.

2. Step 2: Identify why pods are not created

   If no pods appear, missing or wrong Kubernetes credentials is the most common cause, blocking pod creation.

3. Final Answer:

   The Kubernetes cluster credentials are missing or incorrect. -> Option A

4. Quick Check:

   No pods + Kubernetes executor = check cluster credentials [OK]

Hint: No pods? Verify Kubernetes credentials first [OK]

Common Mistakes:

• Blaming DAG syntax errors
• Assuming scheduler executor type is unrelated
• Thinking BashOperator command causes pod creation failure

5. You want to optimize Airflow with Kubernetes executor to handle many short tasks efficiently. Which approach helps reduce pod startup overhead?

hard

A. Increase task retries to avoid pod creation failures.

B. Use pod templates with pre-configured settings and enable pod reuse.

C. Disable Kubernetes executor and switch to SequentialExecutor.

D. Run all tasks in a single pod to save resources.

Solution

1. Step 1: Understand pod startup overhead in Kubernetes executor

   Each pod creation takes time, so reducing overhead improves efficiency for many short tasks.

2. Step 2: Identify optimization methods

   Using pod templates with pre-set configs and enabling pod reuse reduces startup time and resource use.

3. Step 3: Evaluate other options

   Disabling Kubernetes executor or running all tasks in one pod loses scaling benefits. Increasing retries does not reduce pod startup time.

4. Final Answer:

   Use pod templates with pre-configured settings and enable pod reuse. -> Option B

5. Quick Check:

   Pod reuse + templates = faster task startup [OK]

Hint: Use pod templates and reuse to speed up pods [OK]

Common Mistakes:

• Switching to SequentialExecutor loses scaling
• Running all tasks in one pod breaks isolation
• Increasing retries does not reduce pod startup