/developing-applications-on-managed-service-for-apache-flink

Resource Optimization Guide

Overview

This guide covers KPU sizing, operator parallelism tuning, Amazon Managed Service for Apache Flink configuration overrides, and checkpoint resource impact. Use it when right-sizing a new Managed Service for Apache Flink application or optimizing an existing one based on CloudWatch metrics.

KPU Sizing Methodology

CRITICAL: Understanding the KPU Resource Model

Each Kinesis Processing Unit (KPU) provides exactly:

• 1 vCPU
• 4 GB memory (3 GiB JVM heap + 1 GiB reserved for native code allocations including RocksDB, network buffers, and framework overhead)
• 50 GB running application storage

Managed Service for Apache Flink allocates KPUs based on two configuration parameters you set at the service level (console or API):

• Parallelism: The total number of parallel task slots for your application (default: 1, default max: 256)
• ParallelismPerKPU: The number of task slots per KPU (default: 1, max: 8)

MSF derives the KPU count from these settings:

Allocated KPUs = Parallelism / ParallelismPerKPU.

Use ParallelismPerKPU = 1 for most workloads. Increase only when the application is CPU-light and memory-light per task slot (e.g., simple filtering or routing jobs), or for applications with blocking operations (e.g., I/O) where higher values lead to fuller utilization of KPU resources. Higher values pack more task slots per KPU, reducing memory/CPU/storage available per slot.

Estimating Needed KPU Count

Start with the highest of these three estimates, then add headroom:

1. CRITICAL: Throughput-based estimate:

base_kpus = (input_record_rate × avg_record_size_bytes × processing_amplification) / throughput_per_kpu

• processing_amplification: ratio of total bytes processed (including intermediate shuffles) to input bytes. Typically 2–4× for jobs with keyBy and windowing.
• throughput_per_kpu: start with 5–10 MB/s per KPU for typical ETL workloads. CPU-intensive transformations (regex, JSON parsing, ML inference) reduce this to 1–3 MB/s.

2. CRITICAL: State-size-based estimate:

base_kpus = total_state_size_gb / usable_memory_per_kpu_gb

• usable_memory_per_kpu_gb: approximately 2–2.5 GB per KPU after JVM overhead and network buffers (out of 3 GiB heap). The 1 GiB native memory is used by RocksDB and framework overhead. With ParallelismPerKPU = 2, usable memory per slot drops to ~1–1.2 GB.

3. CRITICAL: Source-parallelism-based estimate:

base_kpus = max(kinesis_shard_count, kafka_partition_count)

Source parallelism should match the partition/shard count. If the source has 16 shards, you generally need at least 16 task slots.

Final KPU count with headroom:

recommended_kpus = max(throughput_estimate, state_estimate, source_estimate) × 1.3

The 1.3× multiplier provides ~30% headroom for checkpoint overhead, traffic spikes, and GC pauses. Round up to the nearest even number for balanced TaskManager allocation.

Auto-Scaling Behavior

Managed Service for Apache Flink's built-in auto-scaling uses fixed rules based on containerCPUUtilization:

• Scale up: When containerCPUUtilization exceeds 75% for 15 consecutive 1-minute datapoints, Managed Service for Apache Flink doubles CurrentParallelism (which increases allocated KPUs).
• Scale down: When containerCPUUtilization stays below 10% for 360 consecutive 1-minute datapoints (6 hours), Managed Service for Apache Flink halves CurrentParallelism. Will never reduce below the configured Parallelism setting.
• During scaling: Application enters AUTOSCALING status with downtime. Only StopApplication with Force=true is valid.
• Auto-scaling reacts only to CPU, not heapMemoryUtilization or backpressure. Scale manually for other metrics.
• Default KPU limit: 64 per application. Request increase via Service Quotas.

For finer-grained control, disable built-in auto-scaling and implement custom scaling using CloudWatch alarms and the UpdateApplication API.

Using CloudWatch Metrics to Determine Scaling Direction

After deployment, use these CloudWatch metrics to validate sizing and adjust:

Scaling decision process:

heapMemoryUtilization graduated thresholds:

• Healthy: ≤ 75% — no action needed
• Scale-up / investigation: > 80% sustained — investigate state size, TTL, and consider adding KPUs (see monitoring-and-metrics.md)
• Critical alarm: > 90% — immediate action required; risk of OOM (see checkpoint-tuning.md for OOM diagnostic steps)

1. High containerCPUUtilization + low heapMemoryUtilization → add KPUs (CPU-bound)
2. High heapMemoryUtilization + low CPU → increase KPUs or request memory override (memory-bound)
3. High backPressuredTimeMsPerSecond → identify bottleneck operator, then scale or optimize
4. Growing millisBehindLatest → add KPUs or optimize processing logic

Operator Parallelism Tuning

When to Override Default Parallelism

Managed Service for Apache Flink sets application-level parallelism through the service console (Parallelism and ParallelismPerKPU parameters). In most cases, all operators inherit the configured Parallelism as their default. Override per-operator parallelism with setParallelism() only when:

• A source operator needs parallelism matched to its partition/shard count
• A CPU-intensive operator needs higher parallelism than the rest of the pipeline
• A lightweight operator (simple filter, static enrichment) can run at lower parallelism to free task slots

Do not set application-level parallelism in code for Managed Service for Apache Flink deployments. Managed Service for Apache Flink service-level settings take precedence.

CRITICAL: Source Parallelism Recommendations

Set source parallelism equal to the number of Kinesis shards or Kafka partitions:

// Kinesis: match shard count
DataStream<Event> events = env
    .fromSource(kinesisSource, watermarkStrategy, "kinesis-source")
    .setParallelism(16)  // Match shard count
    .uid("kinesis-source-uid");

If parallelism < partition/shard count, some subtasks handle multiple partitions (uneven load). If parallelism > count, excess subtasks sit idle and waste task slots.

Per-Operator Parallelism and KPU Interaction

When you set per-operator parallelism, Managed Service for Apache Flink still allocates task slots based on the maximum parallelism across all operators. Operators with lower parallelism use fewer slots; operators with higher parallelism require enough total slots to accommodate them.

Example: Parallelism = 16, ParallelismPerKPU = 1 → 16 KPUs, 16 task slots.

// Source: 8 shards → parallelism 8
DataStream<Event> events = env
    .fromSource(kinesisSource, watermarkStrategy, "kinesis-source")
    .setParallelism(8)
    .uid("kinesis-source-uid");

// CPU-heavy processing: inherits default parallelism of 16
DataStream<Result> results = events
    .keyBy(Event::getKey)
    .process(new HeavyProcessor())
    .uid("heavy-processor-uid");

results.sinkTo(sink).uid("sink-uid");

Managed Service for Apache Flink Configuration Overrides

Parameters Overridable via AWS Support

Managed Service for Apache Flink manages most infrastructure configuration automatically. However, certain parameters can be overridden by opening an AWS support case. These include:

Process for Requesting Overrides

1. Gather diagnostic evidence: CloudWatch metrics showing the resource constraint (heap utilization, GC time, checkpoint duration trends)
2. Open an AWS support case under "Managed Service for Apache Flink"
3. Include: application ARN, current KPU count, the specific parameter to override, the requested value, and the diagnostic evidence
4. AWS support applies the override at the service level — no application code changes needed
5. After the override is applied, Managed Service for Apache Flink restarts the application to pick up the new configuration

For checkpoint impact on resources (checkpoint size and memory, frequency vs CPU/network, duration exceeding interval, OOM/GC diagnostic steps), see checkpoint-tuning.md.