AWS CloudWatch for Monitoring Data Pipelines

A pipeline can fail in three ways, and most monitoring catches only the first.

1. It fails loudly. The job errors and stops.
2. It does not run. The schedule was disabled, the trigger never fired, an upstream file never arrived. Nothing fails, because nothing happens.
3. It succeeds with bad data. The job finishes green and loads half the rows, or loads garbage.

The second and third are the ones that reach a dashboard and a director before they reach you. Amazon CloudWatch, which monitors AWS resources and applications with metrics, alarms, dashboards and logs (CloudWatch docs: what is CloudWatch), can catch all three, but only if you tell it what a healthy run looks like.

Metrics: what to measure

Services publish operational metrics on their own. Glue sends job metrics to CloudWatch, Lambda reports invocations, errors and duration, and so on. Those describe the machinery. They do not describe the data.

The metrics that matter most are the ones only your pipeline knows, published as custom metrics:

A few concepts frame every metric (CloudWatch docs: metrics concepts): a namespace is a container that keeps one application's metrics apart from another's; dimensions are name and value pairs, such as the pipeline name, that distinguish series; and a statistic such as sum or maximum is computed over a period.

A worked example

The script publishes four metrics for three nightly runs, defines two alarms, and reads one metric back. The first night's numbers are from a real run of the pipeline in building an ETL pipeline on AWS; the second and third nights are invented, the third to show a failure. It runs against moto, a local emulator, which stores metrics and alarm definitions but does not evaluate alarms.

import datetime as dt
import os
import boto3
from moto import mock_aws

os.environ.update(AWS_ACCESS_KEY_ID="testing", AWS_SECRET_ACCESS_KEY="testing",
                  AWS_DEFAULT_REGION="us-east-1")

# What the last three nightly runs of a pipeline reported (the third is the bad night)
RUNS = [
    {"day": 1, "rows_loaded": 2_870_115, "rows_rejected": 94_509, "minutes": 14.2, "succeeded": 1},
    {"day": 2, "rows_loaded": 2_901_337, "rows_rejected": 96_020, "minutes": 14.9, "succeeded": 1},
    {"day": 3, "rows_loaded": 1_204_556, "rows_rejected": 1_790_113, "minutes": 9.1, "succeeded": 1},
]
DIMENSIONS = [{"Name": "Pipeline", "Value": "trips-to-curated"}]


@mock_aws
def main():
    cw = boto3.client("cloudwatch")
    today = dt.datetime.now(dt.timezone.utc).replace(hour=2, minute=0, second=0, microsecond=0)

    # 1. Each run publishes its own business metrics
    for run in RUNS:
        when = today - dt.timedelta(days=3 - run["day"])
        share = 100 * run["rows_rejected"] / (run["rows_loaded"] + run["rows_rejected"])
        cw.put_metric_data(Namespace="DataPipelines", MetricData=[
            {"MetricName": "RowsLoaded", "Dimensions": DIMENSIONS, "Timestamp": when, "Value": run["rows_loaded"], "Unit": "Count"},
            {"MetricName": "RejectedPercent", "Dimensions": DIMENSIONS, "Timestamp": when, "Value": share, "Unit": "Percent"},
            {"MetricName": "DurationMinutes", "Dimensions": DIMENSIONS, "Timestamp": when, "Value": run["minutes"], "Unit": "None"},
            {"MetricName": "RunSucceeded", "Dimensions": DIMENSIONS, "Timestamp": when, "Value": run["succeeded"], "Unit": "Count"},
        ])

    # 2. Alarms: one on data quality, one on the run not happening at all
    cw.put_metric_alarm(
        AlarmName="trips-rejected-share-high", Namespace="DataPipelines", MetricName="RejectedPercent",
        Dimensions=DIMENSIONS, Statistic="Maximum", Period=86400, EvaluationPeriods=1, DatapointsToAlarm=1,
        Threshold=10.0, ComparisonOperator="GreaterThanThreshold", TreatMissingData="notBreaching",
        AlarmActions=["arn:aws:sns:us-east-1:123456789012:pipeline-alerts"])
    cw.put_metric_alarm(
        AlarmName="trips-no-successful-run", Namespace="DataPipelines", MetricName="RunSucceeded",
        Dimensions=DIMENSIONS, Statistic="Sum", Period=86400, EvaluationPeriods=1, DatapointsToAlarm=1,
        Threshold=1.0, ComparisonOperator="LessThanThreshold", TreatMissingData="breaching",
        AlarmActions=["arn:aws:sns:us-east-1:123456789012:pipeline-alerts"])

    # 3. Read back what a dashboard or an investigation would see
    stats = cw.get_metric_statistics(
        Namespace="DataPipelines", MetricName="RejectedPercent", Dimensions=DIMENSIONS,
        StartTime=today - dt.timedelta(days=3), EndTime=today + dt.timedelta(hours=1),
        Period=86400, Statistics=["Maximum"])["Datapoints"]
    for point in sorted(stats, key=lambda p: p["Timestamp"]):
        flag = "  <-- above the 10% threshold" if point["Maximum"] > 10 else ""
        print(f"rejected share, night {sorted(stats, key=lambda p: p['Timestamp']).index(point) + 1}: {point['Maximum']:5.1f}%{flag}")
    print()
    for alarm in cw.describe_alarms()["MetricAlarms"]:
        print(f"{alarm['AlarmName']:28} {alarm['MetricName']:16} {alarm['ComparisonOperator']:22} {alarm['Threshold']:5}  "
              f"missing data: {alarm['TreatMissingData']}")


main()

rejected share, night 1:   3.2%
rejected share, night 2:   3.2%
rejected share, night 3:  59.8%  <-- above the 10% threshold

trips-rejected-share-high    RejectedPercent  GreaterThanThreshold    10.0  missing data: notBreaching
trips-no-successful-run      RunSucceeded     LessThanThreshold        1.0  missing data: breaching

On the third night the job succeeded. It also rejected 59.8 percent of its input, against about 3 percent normally. A status check sees a green run. The rejected-share metric sees the problem, and the alarm on it, with a threshold of 10 percent, is what would page someone.

Alarms: three settings and one trap

A metric alarm is in one of three states: OK, ALARM or INSUFFICIENT_DATA. Three settings decide when it changes (CloudWatch docs: alarm evaluation):

• Period: the length of time each data point covers.
• Evaluation periods: how many recent data points to look at.
• Datapoints to alarm: how many of those must breach the threshold.

For a nightly batch job the natural period is a day, with one evaluation period. For a streaming consumer, shorter periods with "3 of 5" style settings avoid alarms on a single blip.

The trap is missing data. A pipeline that does not run publishes nothing, and by default an alarm with no data does not go into ALARM. CloudWatch lets you choose how missing data points are treated: as not breaching, as breaching, ignored, or as missing (CloudWatch docs: missing data).

The two alarms in the example use opposite settings on purpose:

• Rejected share: missing data is notBreaching. No run means no quality problem to report here.
• No successful run: missing data is breaching. If no "run succeeded" data point arrives in a day, that absence is the failure, and the alarm fires.

That second alarm is the one that catches a pipeline which silently stopped. It is the most valuable alarm most teams do not have.

Job state: events, not polling

For loud failures, the service will tell you. Glue emits a "Glue Job State Change" event to Amazon EventBridge when a job run reaches SUCCEEDED, FAILED, TIMEOUT or STOPPED, and a rule can route that to an SNS topic, a Lambda function or a Step Functions state machine (Glue docs: automating with EventBridge).

One EventBridge rule matching failed and timed-out states for all jobs, sending to one alert topic, gives baseline coverage for every Glue job in the account, including ones added later. If the pipeline runs under Step Functions, the same idea applies to failed executions; see AWS Step Functions.

Logs: for the question "why?"

Metrics and alarms tell you that something is wrong. Logs tell you why. Glue jobs, Lambda functions and most other services write to CloudWatch Logs, organised into log groups, each with its own retention setting.

Three habits make logs useful:

• Log in a structured form. One JSON object per line, with the run date, the pipeline name and the step. Structured logs can be filtered and aggregated; free text can only be searched.
• Log the numbers. Row counts in and out of each step, and the rejected count per rule.
• Set retention. Decide how long each log group is kept; unlimited retention is a cost that only grows.

CloudWatch Logs Insights then lets you search and analyse log data interactively (CloudWatch docs: Logs Insights): find the first error in last night's run, or count warnings by step over a month.

A monitoring baseline

For each pipeline:

1. A failure notification from job-state events.
2. A "did not run" alarm on a success metric, with missing data treated as breaching.
3. A row-count or rejected-share alarm, with a threshold agreed with the people who use the data.
4. A freshness check on the output table.
5. Structured logs with a retention period.
6. One dashboard showing the last runs: duration, rows, rejections.

Two cautions. Alarms that fire often and mean nothing are soon ignored; tune thresholds until an alarm means "someone must look". And monitoring has a price: the Glue guide notes that enabling job metrics can incur additional CloudWatch charges, and custom metrics, alarms and stored logs are all billed. Measure what you will act on.

Monitoring tells you a problem exists. Preventing bad data from loading in the first place is the job of validation inside the pipeline; data quality testing and validation covers the tests, and AWS Glue jobs the job they run in.

Pipeline monitoring and data quality are covered in the Data Engineering course.

Sources

• What is Amazon CloudWatch? (Amazon CloudWatch User Guide) — AWS Documentation
• Metrics concepts (Amazon CloudWatch User Guide) — AWS Documentation
• Alarm evaluation (Amazon CloudWatch User Guide) — AWS Documentation
• Configuring how CloudWatch alarms treat missing data (Amazon CloudWatch User Guide) — AWS Documentation
• Automating AWS Glue with EventBridge (AWS Glue User Guide) — AWS Documentation
• Monitoring AWS Glue using Amazon CloudWatch metrics (AWS Glue User Guide) — AWS Documentation
• Analyzing log data with CloudWatch Logs Insights (Amazon CloudWatch Logs) — AWS Documentation