Case Study 1 (Part 1): AWS Serverless Ingestion & REST API Trigger

1. Ingestion Architecture Diagram

The first part of our enterprise data pipeline focuses on capturing file arrival events in AWS and initiating execution in Apache Airflow. Notice how the architecture cleanly separates raw file storage from metadata triggers:

graph TD
    subgraph "AWS Object Storage Layer"
        S3_RAW["Amazon S3 Landing Zone<br/>(Raw CSV / Parquet Data Files)"]
        S3_META["Amazon S3 Manifest<br/>(*metadata.json marker file)"]
    end

    subgraph "AWS Serverless Event Layer"
        EVT["AWS EventBridge / S3 Notification<br/>(Filter: ObjectCreated: *metadata.json)"]
        LAM["AWS Lambda Trigger Function<br/>(Manifest Validator & REST Client)"]
    end

    subgraph "Apache Airflow Control Plane"
        API["Airflow REST API Endpoint<br/>POST /api/v1/dags/{dag_id}/dagRuns"]
        SCH["Airflow Scheduler & Webserver"]
    end

    S3_RAW -->|"1. Upload Batch Files"| S3_META
    S3_META -->|"2. Trigger Event Notification"| EVT
    EVT -->|"3. Invoke Serverless Handler"| LAM
    LAM -->|"4. Parse Manifest & Validate Schema"| LAM
    LAM -->|"5. HTTP POST with Runtime Config (conf)"| API
    API -->|"6. Instantiate Pipeline Execution"| SCH

2. Why We Use a Manifest File (metadata.json)

A common mistake in beginner data pipelines is configuring an S3 event trigger to fire immediately whenever any data file (such as part-00001.csv) is uploaded. In production environments, data batches often consist of hundreds of partitioned files written sequentially by upstream systems over several minutes.

If AWS Lambda triggered an Airflow DAG for every individual file:

1. Race Conditions: Airflow would launch dozens of concurrent pipeline runs against partially uploaded datasets.
2. Resource Exhaustion: Each run would attempt to provision its own compute resources, exceeding cloud quotas and escalating infrastructure costs.

The Manifest Pattern

To guarantee consistency, upstream systems follow the manifest design pattern:

1. All raw data files for a given batch are uploaded into an S3 directory (for example, s3://landing-lake/sales/2026-07-07/).
2. Only after the final data file is written does the upstream producer upload a marker file named metadata.json into the exact same directory.
3. Our AWS EventBridge notification rule filters specifically for object keys ending with metadata.json. This guarantees that Airflow is triggered only once per batch, after the dataset is 100% complete and ready for processing.

3. Step-by-Step Code Construction: AWS Lambda Handler

Instead of deploying monolithic scripts, let us break down the AWS Lambda trigger implementation into logical engineering steps. Each snippet explains the exact responsibilities of that component.

Step 3.1: Importing Standard Libraries and Environment Configuration

To minimize Lambda deployment package sizes and avoid managing external layers, we rely on Python standard library modules (urllib.request) for HTTP communications alongside the standard AWS SDK (boto3).

# lambda function.py — Step 1: Imports and Environment Setup
import json
import os
import urllib.request
import urllib.error
import boto3

# Initialize standard AWS S3 client
s3_client = boto3.client('s3')

# Retrieve Airflow endpoint and authentication parameters from environment variables
AIRFLOW_REST_API_URL = os.environ.get("AIRFLOW_REST_API_URL", "https://airflow.internal.company.com/api/v1")
AIRFLOW_DAG_ID = os.environ.get("AIRFLOW_DAG_ID", "etl_emr_event_driven_pipeline")
AIRFLOW_USER = os.environ.get("AIRFLOW_USER", "admin")
AIRFLOW_PASSWORD = os.environ.get("AIRFLOW_PASSWORD", "secret_token")

Step 3.2: Extracting S3 Event Metadata and Validating Object Keys

When Lambda is invoked, it receives an event payload containing bucket and object details. We validate the object key to ensure no stray files trigger the pipeline.

# lambda function.py — Step 2: Event Extraction and Key Validation
def lambda_handler(event, context):
    print(f"Received S3 Notification Event: {json.dumps(event)}")

    # Extract bucket name and object key from the S3 event record
    record = event['Records'][0]
    bucket_name = record['s3']['bucket']['name']
    object_key = record['s3']['object']['key']

    # Defensive check: verify that the triggered file is strictly our manifest marker
    if not object_key.endswith("metadata.json"):
        print(f"Ignoring non-manifest file: {object_key}")
        return {
            "statusCode": 200, 
            "body": "Ignored: Object is not a metadata.json manifest marker."
        }

    print(f"Validated trigger file. Reading manifest from s3://{bucket_name}/{object_key}...")

Step 3.3: Reading the Manifest and Formatting the Airflow Payload

We download metadata.json from S3 and extract key operational variables. These variables are structured into the conf dictionary, which Airflow injects into the DAG execution context as dag_run.conf.

    # Step 3: Fetch and parse manifest contents from S3
    response = s3_client.get_object(Bucket=bucket_name, Key=object_key)
    metadata_content = json.loads(response['Body'].read().decode('utf-8'))

    # Construct target Airflow REST API endpoint URL
    api_endpoint = f"{AIRFLOW_REST_API_URL}/dags/{AIRFLOW_DAG_ID}/dagRuns"

    # Map manifest fields into Airflow runtime configuration dictionary
    payload = {
        "conf": {
            "s3_bucket": bucket_name,
            "metadata_key": object_key,
            "batch_id": metadata_content.get("batch_id", "unknown_batch"),
            "data_path": metadata_content.get("data_path"),
            "target_destinations": metadata_content.get("target_destinations", ["snowflake"]),
            "notify_email": metadata_content.get("notify_email", "data-ops@company.com")
        }
    }
    print(f"Constructed Airflow runtime payload: {json.dumps(payload)}")

Step 3.4: Invoking the Airflow REST API via HTTP POST

We format required authentication headers and execute an HTTP POST request to initiate the DAG run. Comprehensive exception handling ensures any network or authentication failures are logged cleanly in AWS CloudWatch.

    # Step 4: Configure HTTP request headers and authentication
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json"
    }

    # Encode Basic Authentication credentials (in production, use OAuth2 Bearer Tokens or AWS Secrets Manager)
    auth_str = f"{AIRFLOW_USER}:{AIRFLOW_PASSWORD}"
    auth_bytes = auth_str.encode('ascii')
    base64_auth = urllib.request.base64.b64encode(auth_bytes).decode('ascii')
    headers["Authorization"] = f"Basic {base64_auth}"

    # Prepare HTTP Request object
    req = urllib.request.Request(
        url=api_endpoint, 
        data=json.dumps(payload).encode('utf-8'), 
        headers=headers, 
        method='POST'
    )

    # Execute REST API call with strict timeout bounds
    try:
        with urllib.request.urlopen(req, timeout=10) as api_response:
            resp_body = json.loads(api_response.read().decode('utf-8'))
            dag_run_id = resp_body.get('dag_run_id')
            print(f"[SUCCESS] Successfully triggered Airflow DAG Run ID: {dag_run_id}")
            return {
                "statusCode": 200,
                "body": json.dumps({
                    "status": "SUCCESS", 
                    "dag_run_id": dag_run_id,
                    "message": "Airflow pipeline triggered successfully from S3 event."
                })
            }
    except urllib.error.HTTPError as e:
        err_msg = f"[FAILED] Airflow REST API invocation failed with HTTP Status {e.code}: {e.read().decode('utf-8')}"
        print(err_msg)
        raise Exception(err_msg)

4. Verification & Local Testing with cURL

During development, data engineers do not need to upload test files to Amazon S3 or trigger AWS Lambda directly to verify DAG integration. You can simulate the exact HTTP POST payload produced by Lambda using curl against your local Airflow development instance:

curl -X POST "http://localhost:8080/api/v1/dags/etl_emr_event_driven_pipeline/dagRuns" \
  -H "Content-Type: application/json" \
  -u "admin:admin" \
  -d '{
    "conf": {
      "s3_bucket": "test-lake-landing",
      "metadata_key": "events/2026-07-07/metadata.json",
      "batch_id": "batch_test_sim_001",
      "data_path": "s3://test-lake-landing/events/2026-07-07/",
      "target_destinations": ["redshift", "snowflake"],
      "notify_email": "dev-verification@company.com"
    }
  }'

When executed, Airflow returns an HTTP 200 OK JSON response containing the newly created dag_run_id. You can immediately observe the DAG entering execution state within the Airflow Web UI Grid View.

5. Next Steps

With the ingestion trigger layer verified, proceed to Part 2 to learn how enterprise engineering teams use Abstract Base Classes to enforce standardization and design lightweight XCom communication schemas:

• Next Module: Part 2: Enterprise Abstract Base Classes and XCom Variable Design
• Return to Overview: Case Study 1 Overview & Table of Contents

This content is reserved for Premium Members.