IoT Fleet Telemetry Pipeline

AWS architecture template · Kinesis stream ingests device telemetry with three parallel Lambda consumers: anomaly detection, device shadow updates, and cold-path S3 archival for Athena analytics.

IoT Fleet Telemetry Pipeline is a free AWS architecture template built with Lambda, Athena, CloudWatch, DynamoDB, IAM, Kinesis, S3 and SNS. Open it in TheArchitect to edit the diagram, estimate monthly cost, and export the CloudFormation or Terraform that deploys it.

Open in TheArchitect All templates

IoT Fleet Telemetry Pipeline architecture diagram

AWS services in this template

Lambda×3
Athena
CloudWatch
DynamoDB
IAM
Kinesis
S3
SNS

Deployment notes

Needs input

Provision IoT Core or device SDK to publish to Kinesis.
Set Kinesis stream ARN on Lambda event source mappings.

Generated CloudFormation — 13 resources

Logical ID	Resource type
TelemetryStreamkinesis	AWS::Kinesis::Stream
AnomalyDetectorlambda	AWS::Lambda::Function
ShadowUpdaterlambda	AWS::Lambda::Function
ColdArchiverlambda	AWS::Lambda::Function
OpsAlertssns	AWS::SNS::Topic
DeviceShadowdynamodb	AWS::DynamoDB::Table
TelemetryLakes3	AWS::S3::Bucket
FleetAnalyticsathena	AWS::Athena::WorkGroup
ConsumerRoleiam	AWS::IAM::Role
IteratorAgeAlarmcloudwatch	AWS::CloudWatch::Alarm
AnomalyDetectorlambdaOpsAlertssnsPermission	AWS::Lambda::Permission
OpsAlertssnsAnomalyDetectorlambdaSubscription	AWS::SNS::Subscription
ColdArchiverlambdaTelemetryLakes3Permission	AWS::Lambda::Permission

View the full CloudFormation template

{
  "AWSTemplateFormatVersion": "2010-09-09",
  "Description": "Generated by TheArchitect — https://www.thecloudarchitect.dev",
  "Metadata": {
    "TheArchitect": {
      "Edges": [
        {
          "From": "TelemetryStreamkinesis",
          "To": "AnomalyDetectorlambda",
          "Label": "enhanced fan-out"
        },
        {
          "From": "TelemetryStreamkinesis",
          "To": "ShadowUpdaterlambda",
          "Label": "enhanced fan-out"
        },
        {
          "From": "TelemetryStreamkinesis",
          "To": "ColdArchiverlambda",
          "Label": "enhanced fan-out"
        },
        {
          "From": "AnomalyDetectorlambda",
          "To": "OpsAlertssns",
          "Label": "publish alert"
        },
        {
          "From": "ShadowUpdaterlambda",
          "To": "DeviceShadowdynamodb",
          "Label": "upsert shadow"
        },
        {
          "From": "ColdArchiverlambda",
          "To": "TelemetryLakes3",
          "Label": "write parquet"
        },
        {
          "From": "TelemetryLakes3",
          "To": "FleetAnalyticsathena",
          "Label": "query source"
        },
        {
          "From": "ConsumerRoleiam",
          "To": "AnomalyDetectorlambda",
          "Label": "execution role"
        },
        {
          "From": "ConsumerRoleiam",
          "To": "ShadowUpdaterlambda",
          "Label": "execution role"
        },
        {
          "From": "ConsumerRoleiam",
          "To": "ColdArchiverlambda",
          "Label": "execution role"
        }
      ]
    }
  },
  "Resources": {
    "TelemetryStreamkinesis": {
      "Type": "AWS::Kinesis::Stream",
      "Properties": {
        "Name": "fleet-telemetry-stream",
        "StreamModeDetails": {
          "StreamMode": "PROVISIONED"
        },
        "ShardCount": 4,
        "RetentionPeriodHours": 48
      }
    },
    "AnomalyDetectorlambda": {
      "Type": "AWS::Lambda::Function",
      "DependsOn": [
        "ConsumerRoleiam"
      ],
      "Properties": {
        "FunctionName": "fleet-anomaly-detector",
        "Runtime": "python3.12",
        "Handler": "anomaly.handler",
        "MemorySize": 512,
        "Timeout": 60,
        "Description": "Detects out-of-range sensor readings and publishes alerts to SNS.",
        "Role": {
          "Fn::GetAtt": [
            "ConsumerRoleiam",
            "Arn"
          ]
        }
      }
    },
    "ShadowUpdaterlambda": {
      "Type": "AWS::Lambda::Function",
      "DependsOn": [
        "ConsumerRoleiam"
      ],
      "Properties": {
        "FunctionName": "fleet-shadow-updater",
        "Runtime": "nodejs20.x",
        "Handler": "shadow.handler",
        "MemorySize": 256,
        "Timeout": 30,
        "Description": "Updates per-device shadow document in DynamoDB with latest telemetry reading and TTL.",
        "Role": {
          "Fn::GetAtt": [
            "ConsumerRoleiam",
            "Arn"
          ]
        }
      }
    },
    "ColdArchiverlambda": {
      "Type": "AWS::Lambda::Function",
      "DependsOn": [
        "ConsumerRoleiam"
      ],
      "Properties": {
        "FunctionName": "fleet-cold-archiver",
        "Runtime": "python3.12",
        "Handler": "archive.handler",
        "MemorySize": 1024,
        "Timeout": 300,
        "Description": "Batches telemetry records and writes Parquet partitions to S3 for Athena analytics.",
        "Role": {
          "Fn::GetAtt": [
            "ConsumerRoleiam",
            "Arn"
          ]
        }
      }
    },
    "OpsAlertssns": {
      "Type": "AWS::SNS::Topic",
      "Properties": {
        "TopicName": "fleet-ops-alerts",
        "DisplayName": "Fleet Operations Alerts"
      }
    },
    "DeviceShadowdynamodb": {
      "Type": "AWS::DynamoDB::Table",
      "Properties": {
        "TableName": "device-shadow",
        "BillingMode": "PAY_PER_REQUEST",
        "AttributeDefinitions": [
          {
            "AttributeName": "id",
            "AttributeType": "S"
          }
        ],
        "KeySchema": [
          {
            "AttributeName": "id",
            "KeyType": "HASH"
          }
        ]
      }
    },
    "TelemetryLakes3": {
      "Type": "AWS::S3::Bucket",
      "Properties": {
        "BucketName": "fleet-telemetry-lake",
        "VersioningConfiguration": {
          "Status": "Enabled"
        },
        "PublicAccessBlockConfiguration": {
          "BlockPublicAcls": true,
          "BlockPublicPolicy": true,
          "IgnorePublicAcls": true,
          "RestrictPublicBuckets": true
        },
        "NotificationConfiguration": {
          "LambdaConfigurations": [
            {
              "Event": "s3:ObjectCreated:*",
              "Function": {
                "Fn::GetAtt": [
                  "ColdArchiverlambda",
                  "Arn"
                ]
              }
            }
          ]
        }
      },
      "DependsOn": [
        "ColdArchiverlambdaTelemetryLakes3Permission"
      ]
    },
    "FleetAnalyticsathena": {
      "Type": "AWS::Athena::WorkGroup",
      "Properties": {
        "Name": "fleet-analytics",
        "Description": "Fleet telemetry analytical queries",
        "State": "ENABLED",
        "WorkGroupConfiguration": {
          "PublishCloudWatchMetricsEnabled": true,
          "BytesScannedCutoffPerQuery": 5368709120,
          "ResultConfiguration": {
            "OutputLocation": "s3://fleet-telemetry-lake/athena-results/"
          }
        }
      }
    },
    "ConsumerRoleiam": {
      "Type": "AWS::IAM::Role",
      "Properties": {
        "RoleName": "fleet-consumer-role",
        "Description": "Lambda execution role for all Kinesis consumer functions",
        "AssumeRolePolicyDocument": {
          "Version": "2012-10-17",
          "Statement": [
            {
              "Effect": "Allow",
              "Principal": {
                "Service": "lambda.amazonaws.com"
              },
              "Action": "sts:AssumeRole"
            }
          ]
        },
        "ManagedPolicyArns": [
          "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
        ]
      }
    },
    "IteratorAgeAlarmcloudwatch": {
      "Type": "AWS::CloudWatch::Alarm",
      "Properties": {
        "AlarmName": "fleet-stream-iterator-age",
        "AlarmDescription": "Kinesis consumer falling behind — IteratorAge > 60 s indicates backpressure",
        "Namespace": "AWS/Kinesis",
        "MetricName": "GetRecords.IteratorAgeMilliseconds",
        "Statistic": "Maximum",
        "Period": 60,
        "EvaluationPeriods": 3,
        "Threshold": 60000,
        "ComparisonOperator": "GreaterThanThreshold",
        "TreatMissingData": "notBreaching"
      }
    },
    "AnomalyDetectorlambdaOpsAlertssnsPermission": {
      "Type": "AWS::Lambda::Permission",
      "DependsOn": [
        "AnomalyDetectorlambda",
        "OpsAlertssns"
      ],
      "Properties": {
        "Action": "lambda:InvokeFunction",
        "FunctionName": {
          "Ref": "AnomalyDetectorlambda"
        },
        "Principal": "sns.amazonaws.com",
        "SourceArn": {
          "Ref": "OpsAlertssns"
        }
      }
    },
    "OpsAlertssnsAnomalyDetectorlambdaSubscription": {
      "Type": "AWS::SNS::Subscription",
      "DependsOn": [
        "AnomalyDetectorlambdaOpsAlertssnsPermission"
      ],
      "Properties": {
        "TopicArn": {
          "Ref": "OpsAlertssns"
        },
        "Protocol": "lambda",
        "Endpoint": {
          "Fn::GetAtt": [
            "AnomalyDetectorlambda",
            "Arn"
          ]
        }
      }
    },
    "ColdArchiverlambdaTelemetryLakes3Permission": {
      "Type": "AWS::Lambda::Permission",
      "DependsOn": [
        "ColdArchiverlambda"
      ],
      "Properties": {
        "Action": "lambda:InvokeFunction",
        "FunctionName": {
          "Ref": "ColdArchiverlambda"
        },
        "Principal": "s3.amazonaws.com",
        "SourceAccount": {
          "Ref": "AWS::AccountId"
        },
        "SourceArn": {
          "Fn::Sub": "arn:${AWS::Partition}:s3:::fleet-telemetry-lake"
        }
      }
    }
  }
}

Generated by TheArchitect from the diagram above. Review values such as names, CIDR ranges and instance sizes before deploying to your account.

IoT Fleet Telemetry Pipeline

AWS services in this template

Deployment notes

Generated CloudFormation — 13 resources

Related templates