Connection management for cloud services in Apache Airflow - Step-by-Step Execution

Process Flow - Connection management for cloud services

Define Connection in Airflow UI

↓

Store Credentials Securely

↓

Use Connection ID in DAG

↓

Airflow Retrieves Connection

↓

Establish Cloud Service Session

↓

Execute Tasks Using Connection

↓

Close Connection After Use

This flow shows how Airflow manages cloud service connections: define, store, use in DAG, connect, run tasks, then close.

Execution Sample

Apache Airflow

from airflow import DAG
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator

with DAG('example_dag') as dag:
    task = BigQueryInsertJobOperator(
        task_id='bq_task',
        configuration={...},
        gcp_conn_id='my_gcp_conn'
    )

This DAG uses a connection ID 'my_gcp_conn' to connect to Google Cloud BigQuery and run a job.

Process Table

Step	Action	Connection ID Used	Airflow Internal State	Result
1	DAG starts execution	N/A	DAG loaded, task ready	Ready to run task
2	Task requests connection	my_gcp_conn	Fetch connection details from Airflow metadata DB	Connection details retrieved
3	Airflow creates client session	my_gcp_conn	Session object created with credentials	Session ready for API calls
4	Task executes cloud API call	my_gcp_conn	Session used to send request	Cloud job started successfully
5	Task completes	my_gcp_conn	Session closed	Resources freed
6	DAG run ends	N/A	All tasks finished	DAG run successful

💡 All tasks completed and connections closed properly

Status Tracker

Variable	Start	After Step 2	After Step 3	After Step 5	Final
connection_details	None	Credentials and config loaded	Same	Same	None (released)
session	None	None	Client session object created	Session closed	None
task_status	Not started	Waiting for connection	Running	Completed	Success

Key Moments - 3 Insights

Why do we use a connection ID instead of hardcoding credentials in the DAG?

Using a connection ID lets Airflow securely store and manage credentials centrally, as shown in execution_table step 2, avoiding exposure in code.

What happens if the connection details are incorrect or missing?

At step 2 in the execution_table, Airflow fails to retrieve valid connection details, causing the task to error out before creating a session.

When is the cloud service session closed?

As seen in step 5, after the task completes, Airflow closes the session to free resources and avoid leaks.

Visual Quiz - 3 Questions

Look at the execution_table, at which step does Airflow create the client session?

A Step 2

B Step 4

C Step 3

D Step 5

💡 Hint

Check the 'Airflow Internal State' column for 'Session object created'

According to variable_tracker, what is the state of 'session' after step 5?

A None (released)

B Client session object created

C Credentials and config loaded

D Running

💡 Hint

Look at the 'session' row under 'After Step 5' column

If the connection ID is changed in the DAG, which step in execution_table is directly affected?

A Step 1

B Step 2

C Step 4

D Step 6

💡 Hint

Step 2 shows fetching connection details using the connection ID

Concept Snapshot

Airflow connection management:
- Define connections in Airflow UI with IDs
- Use connection ID in DAG operators
- Airflow fetches credentials at runtime
- Creates and uses session for cloud API calls
- Closes session after task completes
- Keeps credentials secure and reusable

Full Transcript

In Airflow, connection management for cloud services means defining connection details like credentials in the Airflow UI. These connections get stored securely in Airflow's metadata database. When a DAG runs, tasks use a connection ID to ask Airflow for the credentials. Airflow then creates a client session to the cloud service using those credentials. The task uses this session to perform cloud operations like starting a BigQuery job. After the task finishes, Airflow closes the session to free resources. This process keeps credentials safe and makes it easy to reuse connections across many DAGs.

Practice

1. What is the main purpose of managing connections in Airflow for cloud services?

easy

A. To manually enter credentials every time a task runs

B. To write custom code for each cloud service login

C. To store cloud login details securely and reuse them in tasks

D. To disable cloud service access in Airflow

Solution

1. Step 1: Understand connection management role

   Connection management stores credentials securely so tasks can access cloud services without exposing secrets.

2. Step 2: Identify reuse benefit

   Reusing connections avoids repeating credentials in multiple tasks, making workflows cleaner and safer.

3. Final Answer:

   To store cloud login details securely and reuse them in tasks -> Option C

4. Quick Check:

   Secure storage + reuse = A [OK]

Hint: Connections store credentials once for many tasks [OK]

Common Mistakes:

• Thinking credentials must be typed in each task
• Assuming connections are only for UI use
• Believing connections disable cloud access

2. Which command correctly adds a Google Cloud connection named 'gcp_default' in Airflow CLI?

easy

A. airflow connections add 'gcp_default' --conn-type google_cloud_platform --conn-extra '{"extra__google_cloud_platform__key_path":"/path/key.json"}'

B. airflow add connection gcp_default --type gcp --extra '{"key_path":"/path/key.json"}'

C. airflow connections create gcp_default --type google --extra '{"key":"/path/key.json"}'

D. airflow connection add gcp_default --conn-type gcp --conn-extra '{"key_path":"/path/key.json"}'

Solution

1. Step 1: Recall correct Airflow CLI syntax

   The command to add connections is 'airflow connections add' followed by connection id and parameters.

2. Step 2: Match correct flags and JSON format

   airflow connections add 'gcp_default' --conn-type google_cloud_platform --conn-extra '{"extra__google_cloud_platform__key_path":"/path/key.json"}' uses '--conn-type google_cloud_platform' and properly formatted JSON for extra parameters, matching Airflow docs.

3. Final Answer:

   airflow connections add 'gcp_default' --conn-type google_cloud_platform --conn-extra '{"extra__google_cloud_platform__key_path":"/path/key.json"}' -> Option A

4. Quick Check:

   Correct command syntax = C [OK]

Hint: Use 'airflow connections add' with correct flags [OK]

Common Mistakes:

• Using 'airflow add connection' instead of 'connections add'
• Wrong JSON key names in --conn-extra
• Missing quotes around JSON string

3. Given this Airflow DAG snippet using a connection ID 'aws_conn', what will happen if the connection is missing?

from airflow.providers.amazon.aws.operators.s3 import S3ListOperator

list_s3 = S3ListOperator(
    task_id='list_s3',
    bucket='my-bucket',
    aws_conn_id='aws_conn'
)

medium

A. The task will create a new connection automatically

B. The task will fail with a connection not found error

C. The task will skip without error

D. The task will run using default AWS credentials

Solution

1. Step 1: Understand connection usage in operator

   The operator uses 'aws_conn_id' to find credentials. If missing, it cannot authenticate.

2. Step 2: Predict behavior on missing connection

   Airflow raises an error when a required connection is not found, causing task failure.

3. Final Answer:

   The task will fail with a connection not found error -> Option B

4. Quick Check:

   Missing connection causes failure = D [OK]

Hint: Missing connection ID causes task failure [OK]

Common Mistakes:

• Assuming default credentials are used automatically
• Thinking task skips if connection missing
• Believing Airflow auto-creates connections

4. You added a connection with the command:

airflow connections add 'azure_default' --conn-type azure --conn-extra '{"tenantId":"abc","clientId":"123"}'

But your tasks fail to authenticate. What is the likely mistake?

medium

A. The JSON keys in --conn-extra are missing the 'extra__azure__' prefix

B. The connection type should be 'azure_cosmos' not 'azure'

C. The connection ID should not be quoted

D. The command should use 'airflow connection add' instead of 'connections add'

Solution

1. Step 1: Check connection type and extra format

   Azure connections require extra keys prefixed with 'extra__azure__' to be recognized properly.

2. Step 2: Identify JSON key format error

   Missing prefix in JSON keys causes Airflow to ignore credentials, leading to authentication failure.

3. Final Answer:

   The JSON keys in --conn-extra are missing the 'extra__azure__' prefix -> Option A

4. Quick Check:

   Correct extra key format fixes auth = B [OK]

Hint: Use 'extra__service__' prefix in --conn-extra keys [OK]

Common Mistakes:

• Using wrong connection type name
• Not prefixing JSON keys correctly
• Wrong CLI command syntax

5. You want to securely manage multiple cloud connections (AWS, GCP, Azure) in Airflow and reuse them in different DAGs. Which approach is best?

hard

A. Use environment variables only and avoid Airflow connections

B. Hardcode credentials in each DAG file for each cloud service

C. Store all credentials in a single connection with mixed cloud keys

D. Create separate connections for each cloud service using Airflow UI or CLI, then reference their IDs in tasks

Solution

1. Step 1: Understand best practice for connection management

   Separate connections per cloud service keep credentials organized and secure.

2. Step 2: Recognize reuse and security benefits

   Referencing connection IDs in tasks avoids duplication and keeps secrets out of code.

3. Final Answer:

   Create separate connections for each cloud service using Airflow UI or CLI, then reference their IDs in tasks -> Option D

4. Quick Check:

   Separate connections + reuse = A [OK]

Hint: Use separate connections per cloud, reference by ID [OK]

Common Mistakes:

• Hardcoding credentials in DAGs
• Mixing all credentials in one connection
• Avoiding Airflow connections for secrets