📊 Data Pipeline Monitoring

Master CloudWatch, CloudTrail, and audit patterns for data pipeline monitoring.

Module: AWS Data Engineering • Topic 29 of 65 • Premium Content

Monitoring Architecture

Architecture Diagram

┌─────────────────────────────────────────────────────────────────────────────┐
│                    DATA PIPELINE MONITORING                                   │
│                                                                             │
│  ┌─────────────────────────────────────────────────────────────────────┐    │
│  │  CLOUDWATCH METRICS                                                  │    │
│  │                                                                     │    │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐             │    │
│  │  │ Glue Jobs    │  │ Kinesis      │  │ Lambda       │             │    │
│  │  │ • DPU Hours  │  │ • IteratorAge│  │ • Invocations│             │    │
│  │  │ • Job Duration│ │ • Throughput │  │ • Duration   │             │    │
│  │  │ • Errors     │  │ • Errors     │  │ • Errors     │             │    │
│  │  └──────────────┘  └──────────────┘  └──────────────┘             │    │
│  └─────────────────────────────────────────────────────────────────────┘    │
│                                                                             │
│  ┌─────────────────────────────────────────────────────────────────────┐    │
│  │  CLOUDWATCH ALARMS                                                   │    │
│  │                                                                     │    │
│  │  ┌───────────────────────────────────────────────────────────────┐  │    │
│  │  │  Alarm: HighIteratorAge                                       │  │    │
│  │  │  Metric: Kinesis.IteratorAgeMilliseconds > 300000 (5 min)    │  │    │
│  │  │  Action: SNS → Email + Lambda → Auto-scale                   │  │    │
│  │  └───────────────────────────────────────────────────────────────┘  │    │
│  │  ┌───────────────────────────────────────────────────────────────┐  │    │
│  │  │  Alarm: GlueJobFailure                                        │  │    │
│  │  │  Metric: Glue.Jobs.Failed > 0                                │  │    │
│  │  │  Action: SNS → Slack + PagerDuty                             │  │    │
│  │  └───────────────────────────────────────────────────────────────┘  │    │
│  └─────────────────────────────────────────────────────────────────────┘    │
│                                                                             │
│  ┌─────────────────────────────────────────────────────────────────────┐    │
│  │  CLOUDTRAIL AUDIT                                                    │    │
│  │                                                                     │    │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐             │    │
│  │  │ S3 Access    │  │ Glue API     │  │ IAM Changes  │             │    │
│  │  │ Logs         │  │ Calls        │  │              │             │    │
│  │  └──────────────┘  └──────────────┘  └──────────────┘             │    │
│  └─────────────────────────────────────────────────────────────────────┘    │
└─────────────────────────────────────────────────────────────────────────────┘

Custom CloudWatch Metrics

import boto3
import time

cloudwatch = boto3.client('cloudwatch')

def put_pipeline_metric(pipeline_name, status, records_processed, duration):
    """Put custom metrics for pipeline monitoring."""
    
    cloudwatch.put_metric_data(
        Namespace='DataPipeline',
        MetricData=[
            {
                'MetricName': 'RecordsProcessed',
                'Dimensions': [
                    {'Name': 'PipelineName', 'Value': pipeline_name},
                    {'Name': 'Status', 'Value': status}
                ],
                'Value': records_processed,
                'Unit': 'Count'
            },
            {
                'MetricName': 'PipelineDuration',
                'Dimensions': [
                    {'Name': 'PipelineName', 'Value': pipeline_name}
                ],
                'Value': duration,
                'Unit': 'Seconds'
            },
            {
                'MetricName': 'PipelineStatus',
                'Dimensions': [
                    {'Name': 'PipelineName', 'Value': pipeline_name}
                ],
                'Value': 1 if status == 'SUCCESS' else 0,
                'Unit': 'None'
            }
        ]
    )

CloudWatch Logs Insights

-- Query Glue job logs
fields @timestamp, @message
| filter @message like /ERROR/
| sort @timestamp desc
| limit 50

-- Query Lambda function logs
fields @timestamp, @duration, @requestId
| stats avg(@duration) as avgDuration, max(@duration) as maxDuration by bin(5m)

-- Query VPC Flow Logs
fields srcAddr, dstAddr, srcPort, dstPort, bytes
| filter action = 'REJECT'
| stats sum(bytes) as totalBytes by srcAddr
| sort totalBytes desc
| limit 20

Interview Q&A

Q1: What CloudWatch metric indicates Kinesis is falling behind?

Answer: IteratorAgeMilliseconds. If this exceeds your SLA threshold, the consumer is not keeping up with the stream.

Q2: How do you monitor Glue job performance?

Answer: Track DPU hours, job duration, rows processed, and error rates. Use CloudWatch Logs for detailed Spark UI logs.

Q3: What is the purpose of CloudTrail in data pipelines?

Answer: CloudTrail records API calls for auditing. Track who accessed what data, when, and from where.

Summary

Metrics: CloudWatch for pipeline-specific metrics
Alarms: Proactive alerting on anomalies
Logs: CloudWatch Logs Insights for log analysis
Audit: CloudTrail for API call tracking
Dashboards: CloudWatch Dashboards for visualization

Data Pipeline Monitoring on AWS