Parallelism and Operator Chains

The parallelism of a specific operator is defined by the number of its subtasks. A data flow containing parallel subtasks is known as a parallel stream, requiring multiple stream partitions to distribute the workload. Generally, the parallelism of a Flink program is determined by the maximum parallelism among all its operators. However, individual operators within the same program can possess distinct parallelism settings.

For instance, consider a pipeline consisting of Source, Map, Window, and Sink operators. If the Sink has a parallelism of 1 while the others are set to 2, the overall program parallelism is considered 2.

Unlike Spark's RDD partitions, Flink allows parallelism configuration at the operator level. The priority for setting parallelism is as follows: Operator-specific setting > Environment setting > Submission command argument > Configuration file.

Operator Chaining

Data transmission between operators occurs in two primary forms: one-to-one (forwarding) or redistributing.

One-to-One Forwarding In this mode, partitioning and element order are preserved. For example, data read by a Source operator can be directly passed to a Map operator without repartitioning. The Map subtasks receive elements in the exact same order and quantity as produced by the Source subtasks. Operators like map, filter, and flatMap typically exhibit this behavior, similar to narrow dependencies in Spark.

Redistributing Here, data partitions change. This occurs betweeen operators such as map and keyBy, or between keyBy and Sink. Subtasks send data to different downstream targets based on transmission strategies, causing a repartitioning effect similar to Spark's shuffle (wide dependency).

Chaining Optimization Flink optimizes performance by linking consecutive one-to-one operators with identical parallelism into a single task. This forms an "Operator Chain." Each task executes within a single thread. For example, if Source and Map both have parallelism 2 and are one-to-one, they merge into one task with two subtasks, reducing thread switching and buffer exchanges.

Chaining is enabled by default but can be controlled programmatically:

// Prevent chaining with previous or next operators
.dataStream.map(event -> event.getId()).disableChaining();

// Start a new chain from this operator
.dataStream.map(event -> event.getId()).startNewChain();

Separating operators is useful when their logic is computationally heavy.

Task Slots and Resource Management

Each TaskManager in Flink runs as a JVM process capable of spawning multiple threads to execute subtasks concurrently. To manage concurrency limits, resources are divided into Task Slots. Each slot represents a fixed subset of resources within a TaskManager dedicated to executing a subtask.

Currently, slots isolate memory but not CPU. It is common practice to configure the number of slots to match the CPU core count to minimize contention.

Slot Sharing By default, subtasks from different operators within the same job can share a slot. For example, if a job has a Source→Map chain (parallelism 6) and a KeyBy→Sink chain (parallelism 1), the subtasks for the second chain can share slots with the first, provided they belong to the same job. This allows resource-intensive and lightweight tasks to coexist, balancing load across TaskManagers. It also ensures job resilience; if one TaskManager fails, others can take over the shared slots.

Slot sharing can be restricted using slot sharing groups:

// Assign to a specific sharing group
.dataStream.map(event -> event.getId()).slotSharingGroup("groupA");

Tasks in different groups are isolated and must occupy distinct slots. The total required slots equal the sum of the maximum parallelism of each group.

Slots vs. Parallelism Task slots are static resources configured via taskmanager.numberOfTaskSlots, representing capacity. Parallelism is dynamic, configured via parallelism.default, representing actual usage during execution. For example, a cluster with 3 TaskManagers each having 3 slots has a total capacity of 9 slots. If a job defines 4 operators with parallelism 2, the required parallelism is 2, consuming 2 slots per TaskManager depending on distribution.

Job Submission Lifecycle

The submission process transforms the user code into an executable graph through several stages:

1. StreamGraph: Generated client-side from the DataStream API code, representing the initial logical DAG.
2. JobGraph: An optimized version of the StreamGraph submitted to the JobManager. Operators are chained into tasks here to minimize overhead.
3. ExecutionGraph: Created by the JobMaster from the JobGraph. This is the parallelized version where tasks are split into subtasks according to parallelism settings.
4. Physical Graph: The actual deployment layout on TaskManagers. It defines data location and transmission details for execution.

DataStream API Overview

The DataStream API is the core abstraction for Flink programs. Since version 1.12, it supports both streaming and batch execution modes (via BATCH setting), making the DataSet API obsolete.

Execution Modes

• Streaming: Default mode for unbounded data.
• Batch: For bounded data processing.
• Automatic: Selects mode based on data source characteristics.

Lazy Evaluation Defining transformations does not trigger execution. Flink is event-driven; computation starts only when env.execute() is called. This method blocks until the job completes. executeAsync() allows non-blocking submission but is less common than defining separate job classes.

Type Information Flink uses TypeInformation to handle serialization and comparison. Supported types include:

• Primitives: Java primitives, String, BigDecimals, etc.
• Arrays: Primitive and object arrays.
• Composite: Tuples (up to 25 fields), POJOs (Java beens), Rows.
• Generics: Handled via Kryo serialization if not matching POJO rules.

POJOs are preferred for readability and key extraction. Requirements include a public class, no-arg constructor, public fields, and serializable types.

Transformation Operators

Consider a stream of TemperatureReading objects.

DataStream<TemperatureReading> inputStream = env.fromElements(
    new TemperatureReading("dev_A", 1000L, 25.0),
    new TemperatureReading("dev_B", 1001L, 30.0)
);

Map Transforms one element in to another.

// Using Lambda
SingleOutputStreamOperator<String> idStream = inputStream.map(sensor -> sensor.getDeviceId());

// Using Anonymous Class
SingleOutputStreamOperator<String> idStream2 = inputStream.map(new MapFunction<TemperatureReading, String>() {
    @Override
    public String map(TemperatureReading value) {
        return value.getDeviceId();
    }
});

Filter Retains elements based on a boolean condition.

inputStream.filter(sensor -> "dev_A".equals(sensor.getDeviceId())).print();

FlatMap Can produce zero, one, or multiple elements per input.

inputStream.flatMap((sensor, out) -> {
    if (sensor.getValue() > 20.0) {
        out.collect("High Temp: " + sensor.getValue());
    }
}).print();

Aggregation Operations

KeyBy Logically partitions the stream by key using hash codes. Ensures identical keys go to the same parallel subtask.

KeyedStream<TemperatureReading, String> keyedStream = inputStream.keyBy(TemperatureReading::getDeviceId);

Simple Aggregations Operations like sum, min, max work on keyed streams. minBy and maxBy return the entire record containing the min/max value, whereas min/max only update the specific field.

Reduce Combines current state with new input.

keyedStream.reduce((s1, s2) -> 
    new TemperatureReading(s1.getDeviceId(), s2.getTimestamp(), s1.getValue() + s2.getValue())
).print();

Rich Functions Rich functions (e.g., RichMapFunction) provide lifecycle methods (open, close) and access to runtime context. open is called once per subtask before processing begins.

Physical Partitioning

Controls how data is distributed downstream.

• Shuffle: Random uniform distribution (stream.shuffle()).
• Rebalance: Round-robin distribution (stream.rebalance()).
• Rescale: Round-robin but restricted to a subset of downstream tasks (stream.rescale()).
• Broadcast: Sends copy to all downstream tasks (stream.broadcast()).
• Global: Sends all data to the first downstream task (stream.global()).
• Custom: User-defined logic via Partitioner.

class CustomPartitioner implements Partitioner<String> {
    public int partition(String key, int numPartitions) {
        return Integer.parseInt(key) % numPartitions;
    }
}
// Usage
stream.partitionCustom(new CustomPartitioner(), value -> value);

Stream Splitting and Side Outputs

Side Outputs Allows splitting a stream into multiple logical streams based on conditions within a ProcessFunction.

OutputTag<TemperatureReading> highTempTag = new OutputTag<>("high-temp") {};

SingleOutputStreamOperator<TemperatureReading> processed = inputStream.process(new ProcessFunction<TemperatureReading, TemperatureReading>() {
    @Override
    public void processElement(TemperatureReading value, Context ctx, Collector<TemperatureReading> out) {
        if (value.getValue() > 50.0) {
            ctx.output(highTempTag, value);
        } else {
            out.collect(value);
        }
    }
});

processed.getSideOutput(highTempTag).print("High Temp Stream");

Connecting Streams

Union Merges multiple streams of the same type.

DataStream<Integer> stream1 = env.fromElements(1, 2);
DataStream<Integer> stream2 = env.fromElements(3, 4);
DataStream<Integer> unioned = stream1.union(stream2);

Connect Connects two streams of potentially different types. Requires a CoMapFunction to handle each type separately.

DataStream<Integer> ints = env.fromElements(1, 2);
DataStream<String> strings = env.fromElements("a", "b");

ConnectedStreams<Integer, String> connected = ints.connect(strings);

SingleOutputStreamOperator<String> result = connected.map(new CoMapFunction<Integer, String, String>() {
    @Override
    public String map1(Integer value) { return value.toString(); }
    @Override
    public String map2(String value) { return value; }
});

Sink Operators

Flink writes results to external systems. For Kafka, the KafkaSink builder is used.

KafkaSink<String> sink = KafkaSink.<String>builder()
    .setBootstrapServers("broker1:9092,broker2:9092")
    .setRecordSerializer(KafkaRecordSerializationSchema.builder()
        .setTopic("output-topic")
        .setValueSerializationSchema(new SimpleStringSchema())
        .build()
    )
    .build();

dataStream.sinkTo(sink);