Performance Tuning

Parallelism Guidelines:

Kafka Source:
  parallelism = number of Kafka partitions
  (each instance reads 1+ partitions; extras sit idle)

CPU-bound operators (map, filter, complex UDFs):
  parallelism = available CPU cores / operators
  Start with source parallelism, increase if busyTime is high

Keyed operators (window, process):
  parallelism ≤ number of distinct keys
  (if 100 keys, parallelism > 100 means idle instances)

Sink operators:
  parallelism = depends on sink capacity
  JDBC: limited by connection pool and DB throughput
  Kafka: can match source parallelism
  Elasticsearch: limited by bulk indexing capacity

Example job tuning:
  Source (Kafka, 24 partitions):  parallelism = 24
  Map/Filter (CPU-light):         parallelism = 24 (chained with source)
  KeyBy + Window (CPU-heavy):     parallelism = 48 (double for heavy compute)
  Sink (JDBC, batched):           parallelism = 12 (DB can't handle more)

Over-parallelism symptoms:
  - Many subtasks with near-zero throughput
  - High network overhead (more shuffling)
  - Checkpoint takes longer (more tasks to coordinate)
  - Wasted memory (each instance has minimum overhead)

# RocksDB State Backend Tuning (flink-conf.yaml)

# Use managed memory for RocksDB (recommended)
state.backend.rocksdb.memory.managed: true

# Block cache: caches frequently-read SST blocks
# Larger = fewer disk reads for state access
state.backend.rocksdb.block.cache-size: 256mb

# Write buffer: buffers writes before flushing to SST files
# Larger = fewer flushes, better write throughput
state.backend.rocksdb.writebuffer.size: 128mb
state.backend.rocksdb.writebuffer.count: 4

# Compaction: merges SST files to reduce read amplification
# Level compaction (default) is best for most workloads
state.backend.rocksdb.compaction.style: LEVEL
state.backend.rocksdb.compaction.level.max-size-level-base: 256mb

# Bloom filters: reduce unnecessary disk reads for point lookups
state.backend.rocksdb.use-bloom-filter: true
state.backend.rocksdb.bloom-filter.bits-per-key: 10

# Incremental checkpoints (critical for large state)
state.backend.incremental: true

# Number of threads for background compaction and flush
state.backend.rocksdb.thread.num: 4

# Network Memory Configuration

# Fraction of total memory for network buffers (default 10%)
taskmanager.memory.network.fraction: 0.1
taskmanager.memory.network.min: 64mb
taskmanager.memory.network.max: 1gb

# Buffer size (default 32KB)
# Larger buffers = higher throughput but higher latency
taskmanager.memory.segment-size: 32kb

# Buffer timeout: max time before flushing partially-full buffer
# Low value = low latency (flush often, even if buffer not full)
# High value = high throughput (wait for full buffers)
execution.buffer-timeout: 100ms  # default

# For low-latency applications:
execution.buffer-timeout: 1ms    # flush almost immediately

# For high-throughput batch-like processing:
execution.buffer-timeout: -1     # only flush when buffer is full

# Exclusive buffers per channel (for credit-based flow control)
taskmanager.network.memory.buffers-per-channel: 2
# Floating buffers shared across channels
taskmanager.network.memory.floating-buffers-per-gate: 8

# Checkpoint Configuration

# Interval: time between checkpoint triggers
execution.checkpointing.interval: 60s
# Too short: constant I/O pressure
# Too long: more data to reprocess on failure

# Minimum pause between checkpoints
execution.checkpointing.min-pause: 30s
# Prevents overlap: next checkpoint waits at least 30s after previous completes

# Timeout: fail checkpoint if not complete within this time
execution.checkpointing.timeout: 600s
# Set higher than your worst-case checkpoint duration

# Tolerate N consecutive failures before job fails
execution.checkpointing.tolerable-failed-checkpoints: 3

# Unaligned checkpoints (reduce checkpoint duration under backpressure)
execution.checkpointing.unaligned.enabled: true
# Only enable if aligned checkpoints are slow due to backpressure

# Externalized checkpoints (keep on cancellation)
execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION

# Max concurrent checkpoints
execution.checkpointing.max-concurrent-checkpoints: 1
# >1 only if checkpoints are very fast and interval is short

# Checkpoint storage
state.checkpoints.dir: s3://bucket/checkpoints/
# Use S3 with high throughput (not single-threaded upload)
fs.s3a.connection.maximum: 100
fs.s3a.threads.max: 64

TaskManager Memory Model:
═══════════════════════════════════════════════════════════
Total Process Memory: 8192 MB (taskmanager.memory.process.size)
├── Total Flink Memory: ~7200 MB
│   ├── Task Heap: ~2880 MB (user code objects)
│   │   └── taskmanager.memory.task.heap.size (or auto-calculated)
│   ├── Managed Memory: ~2880 MB (40% of Flink memory)
│   │   └── RocksDB, batch sorting, Python UDFs
│   │   └── taskmanager.memory.managed.fraction: 0.4
│   ├── Network Memory: ~720 MB (10% of Flink memory)
│   │   └── Shuffle buffers between operators
│   │   └── taskmanager.memory.network.fraction: 0.1
│   └── Framework Heap: 128 MB (Flink internals)
│       └── taskmanager.memory.framework.heap.size: 128mb
├── JVM Metaspace: 256 MB
│   └── taskmanager.memory.jvm-metaspace.size: 256mb
└── JVM Overhead: ~820 MB (10% of total)
    └── GC, threads, native memory
    └── taskmanager.memory.jvm-overhead.fraction: 0.1
═══════════════════════════════════════════════════════════

Sizing guidelines:
  - Heavy UDFs (complex objects): increase task heap
  - Large RocksDB state: increase managed memory fraction
  - High parallelism (many channels): increase network memory
  - Many classes/libraries: increase metaspace

// ✅ Good: Use Flink-native types (fastest serialization)
DataStream<Tuple3<String, Long, Double>> stream = ...;
// Flink knows exact layout, no reflection, minimal bytes

// ✅ Good: POJO (fast, supports evolution)
// Requirements: public no-arg constructor, public fields or getters/setters
public class OrderEvent {
    public String orderId;
    public long timestamp;
    public double amount;
    public OrderEvent() {} // no-arg constructor required!
}

// ❌ Bad: Types that fall back to Kryo
// - Classes without no-arg constructor
// - Scala case classes (without TypeInformation)
// - Generic types without TypeHint
// - Third-party library classes

// Check for Kryo fallback in logs:
// WARN: "Class X is not a valid POJO, falling back to Kryo"

// Force Flink to fail instead of falling back to Kryo:
env.getConfig().disableGenericTypes();
// Now any type that would use Kryo causes a compile error

// Register custom serializer for third-party types:
env.getConfig().registerTypeWithKryoSerializer(
    ThirdPartyClass.class, ThirdPartySerializer.class);