Key Takeaways

  • Kinesis Data Streams captures high-throughput IoT sensor data; Apache Flink on EMR processes it with sub-second latency using stateful stream processing.
  • Windowed aggregations (tumbling, sliding, session) let you compute rolling metrics — average temperature per device per minute — without storing raw events in a database first.
  • Flink's KeyedProcessFunction and state backends (RocksDB) let you maintain per-device state across millions of events without external lookups.
  • ML inference can run inline in the Flink pipeline by calling a SageMaker endpoint per record or per micro-batch, enabling real-time anomaly detection.
  • EMR on EC2 or EMR Serverless both run Flink; EMR Serverless removes cluster management for variable IoT traffic patterns.

IoT sensor fleets generate continuous, high-volume event streams — temperature, vibration, GPS, device health — that need processing within seconds, not after a nightly batch job. Kinesis Data Streams is the durable, ordered ingestion layer; Apache Flink is the processing engine that can hold state (like a running average per device) across unbounded streams, which batch frameworks like Spark struggle to do efficiently at low latency.

Running Flink on Amazon EMR gives you a managed Flink runtime with direct integration to the Kinesis Connector, S3 for checkpointing, and SageMaker for inference — without hand-rolling a Flink cluster on EC2.

Architecture Overview

IoT Devices → Kinesis Data Streams → Flink on EMR → Windowed Aggregation
                                           │
                                           ├──► SageMaker Endpoint (anomaly inference)
                                           │
                                           └──► S3 / OpenSearch (dashboards, alerts)

Step 1: Create the Kinesis Stream

aws kinesis create-stream \
  --stream-name iot-sensor-events \
  --shard-count 4

Shard count determines ingest throughput — each shard supports 1,000 records/sec or 1MB/sec write. For a fleet of 50,000 devices reporting every 10 seconds, 4-6 shards is a reasonable starting point; scale with Kinesis on-demand mode if traffic is unpredictable.

aws kinesis update-shard-count \
  --stream-name iot-sensor-events \
  --target-shard-count 8 \
  --scaling-type UNIFORM_SCALING
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

Properties consumerConfig = new Properties();
consumerConfig.put(AWSConfigConstants.AWS_REGION, "me-south-1");
consumerConfig.put(ConsumerConfigConstants.STREAM_INITIAL_POSITION, "LATEST");

DataStream<SensorEvent> events = env.addSource(
    new FlinkKinesisConsumer<>(
        "iot-sensor-events",
        new SensorEventDeserializer(),
        consumerConfig
    )
).assignTimestampsAndWatermarks(
    WatermarkStrategy.<SensorEvent>forBoundedOutOfOrderness(Duration.ofSeconds(5))
        .withTimestampAssigner((event, ts) -> event.getEventTimeMillis())
);

Step 3: Windowed Aggregation Per Device

DataStream<DeviceAverage> averages = events
    .keyBy(SensorEvent::getDeviceId)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .aggregate(new AverageAggregateFunction());

A tumbling window of 1 minute produces non-overlapping, fixed-size aggregates — one average temperature reading per device per minute. For smoother trend detection, a sliding window (e.g., 5-minute window, 1-minute slide) gives overlapping aggregates updated every minute.

DataStream<DeviceAverage> slidingAverages = events
    .keyBy(SensorEvent::getDeviceId)
    .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1)))
    .aggregate(new AverageAggregateFunction());

Step 4: Calling SageMaker for Real-Time Anomaly Detection

public class AnomalyInferenceFunction extends RichAsyncFunction<DeviceAverage, ScoredEvent> {
    private transient SageMakerRuntimeAsyncClient client;

    @Override
    public void open(Configuration parameters) {
        client = SageMakerRuntimeAsyncClient.builder()
            .region(Region.ME_SOUTH_1)
            .build();
    }

    @Override
    public void asyncInvoke(DeviceAverage input, ResultFuture<ScoredEvent> resultFuture) {
        InvokeEndpointRequest request = InvokeEndpointRequest.builder()
            .endpointName("iot-anomaly-detector")
            .contentType("application/json")
            .body(SdkBytes.fromUtf8String(input.toJson()))
            .build();

        client.invokeEndpoint(request).thenAccept(response -> {
            double score = parseScore(response.body().asUtf8String());
            resultFuture.complete(Collections.singleton(new ScoredEvent(input, score)));
        });
    }
}

Using Flink's async I/O pattern (RichAsyncFunction) prevents the SageMaker call from blocking the stream — critical, since synchronous calls per-record would collapse throughput at IoT scale.

Step 5: Deploying on EMR

aws emr create-cluster \
  --name "iot-flink-cluster" \
  --release-label emr-7.1.0 \
  --applications Name=Flink \
  --instance-type m6g.xlarge \
  --instance-count 4 \
  --use-default-roles \
  --region me-south-1

For bursty IoT traffic (daytime spikes, overnight lulls), EMR Serverless avoids paying for idle cluster capacity:

aws emr-serverless create-application \
  --name iot-flink-app \
  --release-label emr-7.1.0 \
  --type FLINK

Checkpointing and Fault Tolerance

Flink checkpoints state to S3 so a job restart doesn't lose in-flight windowed aggregates:

env.enableCheckpointing(30000); // every 30 seconds
env.setStateBackend(new EmbeddedRocksDBStateBackend());
env.getCheckpointConfig().setCheckpointStorage("s3://f9-flink-checkpoints/iot-pipeline/");

RocksDB as the state backend handles state sizes beyond what fits in memory — important when you're tracking per-device state for hundreds of thousands of devices simultaneously.

Pipeline Latency Breakdown

Kinesis ingest ~50ms Flink window ~200ms SageMaker inference ~300ms Alert/sink ~80ms

FAQ

Why use Flink instead of Spark Structured Streaming for this? Flink's native support for low-latency, event-time windowing and stateful processing (via RocksDB) generally outperforms Spark's micro-batch model for true streaming use cases like per-device rolling aggregates under one second.

Does EMR Serverless support Flink? Yes, as of EMR release 6.x/7.x, EMR Serverless supports Flink applications alongside Spark and Hive, billing only for actual compute used.

How do I handle late-arriving IoT events (devices with unreliable connectivity)? Configure allowedLateness on your window and use a watermark strategy with bounded out-of-orderness; events beyond that are routed to a side-output for separate handling.

Can this pipeline write directly to a dashboard? Yes — sink aggregated results to OpenSearch or a DynamoDB table backing a real-time dashboard, in addition to or instead of S3 for batch analytics.

Build Your Real-Time IoT Pipeline

F9infotech designs and operates Kinesis + Flink pipelines for IoT and event-driven workloads at scale, including ML inference integration. Contact our data engineering team to scope your streaming architecture.