Forwarding vs redistributing data in Flink | Data Engineering

Q: How is forwarding different from redistributing?

// Forwarding: chained into one thread stream.filter(txn txn.isValid()) // forwarding .map(txn txn.enrich()); // forwarding // Cost: zero network, zero serialization // Redistributing: shuffle boundary stream.keyBy(Transaction::getAccountId) // redistributing .sum("amount"); // keyed aggregation //

Definition

Forwarding keeps records on the same partition in order with no shuffle, while redistributing shuffles records across subtasks by a rule like key hashing at a network cost.

Forwarding edge – A one-to-one connection where each upstream subtask sends records directly to exactly one downstream subtask, keeping records in the same partition, preserving order, and requiring no network shuffle.
Operator chaining – When consecutive operators are connected by forwarding edges with the same parallelism, Flink fuses them into a single task chain running in one thread with zero serialization.
Redistributing edge – A many-to-many connection where records are repartitioned across all downstream subtasks according to a rule, involving serialization, network transfer, and deserialization.
keyBy – The most common redistributing operation, which hashes each record's key and routes it to the subtask responsible for that hash bucket.
Shuffle boundary – The point in the pipeline where redistribution occurs, which is the most expensive operation in a streaming job.

Imagine the National Bank of India has three clerks reading transaction slips and three accountants recording them. Forwarding is like clerk one always handing to accountant one, clerk two to accountant two, and clerk three to accountant three: each slip stays on its own conveyor belt, in order, with no shuffling. Redistributing is like requiring that all slips for the same account go to the same accountant regardless of which clerk read them, so slips get sorted and shuffled across belts, which costs time and mixing but guarantees that all of Rajesh's slips land together.

Understanding the Concepts

What is a forwarding edge in Flink?

A forwarding edge appears when consecutive operators have the same parallelism and you have not introduced any repartitioning operation. Each parallel instance of the upstream operator feeds exactly one parallel instance of the downstream operator. Records never cross the network.

JAVA

package com.bank.streaming;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class ForwardingDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // Source -> Filter -> Map: all forwarding edges
        // All run at parallelism 4 with no repartitioning
        env.addSource(new KafkaSource<>("nbi-transactions"))
            .filter(txn -> txn.getAmount()
                .compareTo(BigDecimal.ZERO) > 0)   // forwarding
            .map(txn -> txn.normalize())             // forwarding
            .print("Clean");

        // Flink chains these into one task: Source+Filter+Map
        // Zero serialization, zero network, one thread per subtask
        env.execute("NBI Forwarding Demo");
    }
}

TEXT

Forwarding chain (parallelism 4):
  Subtask-0: Source-0 -> Filter-0 -> Map-0  (one thread)
  Subtask-1: Source-1 -> Filter-1 -> Map-1  (one thread)
  Subtask-2: Source-2 -> Filter-2 -> Map-2  (one thread)
  Subtask-3: Source-3 -> Filter-3 -> Map-3  (one thread)
  No network shuffle between operators

What is a redistributing edge in Flink?

A redistributing edge appears whenever records must be repartitioned. The most common cause is keyBy, which hashes each record's key and routes it to the downstream subtask responsible for that hash bucket. Redistribution requires serializing each record, sending it over the network, and deserializing it.

JAVA

package com.bank.streaming;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class RedistributingDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        env.addSource(new KafkaSource<>("nbi-transactions"))
            .map(txn -> txn.normalize())  // forwarding
            // keyBy: REDISTRIBUTING edge
            // Each transaction hashed by accountId to correct subtask
            .keyBy(Transaction::getAccountId)
            .sum("amount") // keyed aggregation
            .print("Running total");

        // All ACC-001 transactions go to same subtask
        // All ACC-002 transactions go to same subtask
        // Network shuffle between map and sum operators
        env.execute("NBI Redistributing Demo");
    }
}

How is forwarding different from redistributing?

DIAGRAM

| Feature | Forwarding Edge | Redistributing Edge |
|---|---|---|
| Connection | One-to-one (subtask to subtask) | Many-to-many (all to all) |
| Network cost | Zero (same thread) | Serialization + network transfer |
| Order | Preserved within partition | Broken across partitions |
| Operator chaining | Yes (fused into one thread) | No (shuffle boundary) |
| Examples | source -> map, filter -> map | keyBy, rebalance, broadcast |
| Performance | Free | Most expensive operation |

JAVA

// Forwarding: chained into one thread
stream.filter(txn -> txn.isValid())   // forwarding
      .map(txn -> txn.enrich());       // forwarding
// Cost: zero network, zero serialization

// Redistributing: shuffle boundary
stream.keyBy(Transaction::getAccountId)  // redistributing
      .sum("amount");                     // keyed aggregation
// Cost: serialize -> network -> deserialize per record

What are the different redistributing strategies?

Flink provides several redistributing operations beyond keyBy. Each serves a different purpose: rebalance for load spreading, rescale for local redistribution, broadcast for sending to all subtasks, and custom partitioning for user-defined routing.

JAVA

package com.bank.streaming;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class RedistributionStrategiesDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        DataStream<Transaction> stream =
            env.addSource(new KafkaSource<>("nbi-transactions"));

        // keyBy: hash by key, all same-key records to same subtask
        stream.keyBy(Transaction::getAccountId)
            .sum("amount");

        // rebalance: round-robin to spread load evenly
        stream.rebalance()
            .map(txn -> txn.enrich());

        // broadcast: every record to every subtask
        // Use for small lookup data (e.g., fraud rules)
        stream.broadcast()
            .connect(rulesStream);

        // rescale: local redistribution (fewer network connections)
        stream.rescale()
            .map(txn -> txn.process());

        env.execute("NBI Redistribution Strategies Demo");
    }
}

How does key skew affect redistribution?

A keyBy on a skewed key distribution can overload one subtask and cripple parallelism. If 80% of transactions belong to one corporate account, one subtask gets 80% of the data while the other subtasks sit nearly idle.

JAVA

package com.bank.streaming;

public class KeySkewDemo {
    // PROBLEM: one corporate account has 80% of traffic
    // keyBy(accountId) sends 80% to one subtask

    // WRONG: naive keyBy on skewed account
    // stream.keyBy(Transaction::getAccountId)
    //     .process(new FraudScorer());
    // Result: subtask-2 gets 80%, others nearly idle

    // SOLUTION 1: salt the key for parallelism
    // stream.keyBy(txn -> txn.getAccountId()
    //     + "-" + (txn.hashCode() % 4)) // 4 sub-keys per account
    //     .process(new FraudScorer());

    // SOLUTION 2: pre-aggregate before keyBy
    // stream.rebalance()           // spread evenly
    //     .map(txn -> preScore(txn)) // partial scoring
    //     .keyBy(Transaction::getAccountId)
    //     .process(new MergeScores()); // merge partial results
}

TEXT

Key skew impact:
  Normal distribution:   Subtask-0: 25%  Subtask-1: 25%  Subtask-2: 25%  Subtask-3: 25%
  Skewed distribution:   Subtask-0:  5%  Subtask-1:  5%  Subtask-2: 80%  Subtask-3: 10%
  Effective parallelism: 4x                            only 1.25x (wasted slots)

How does operator chaining optimize forwarding edges?

When consecutive operators are connected by forwarding edges, Flink fuses them into a single task chain that runs in one thread. This eliminates serialization, deserialization, and network transfer entirely, making chained operations essentially free.

JAVA

package com.bank.streaming;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class ChainingDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        env.addSource(new KafkaSource<>("nbi-transactions"))
            .filter(txn -> txn.isValid())      // chained with source
            .map(txn -> txn.enrich())           // chained
            .map(txn -> txn.normalize())        // chained
            // All above run in ONE thread per subtask
            .keyBy(Transaction::getAccountId)   // SHUFFLE: breaks chain
            .process(new FraudScorer())         // new chain starts
            .addSink(new AlertSink());          // chained with scorer

        env.execute("NBI Chaining Demo");
    }
}

What are common mistakes with forwarding and redistributing?

The most common mistake is placing expensive filtering or projection after a keyBy when it could be done before, causing the shuffle to carry more data than necessary. Another mistake is not monitoring for key skew after a keyBy.

JAVA

// WRONG: filter after keyBy (shuffles unnecessary data)
stream
    .keyBy(Transaction::getAccountId)  // shuffles ALL records
    .filter(txn -> txn.getAmount()
        .compareTo(new BigDecimal("100")) > 0) // too late!
    .sum("amount");

// CORRECT: filter before keyBy (shuffles less data)
stream
    .filter(txn -> txn.getAmount()
        .compareTo(new BigDecimal("100")) > 0) // filter first
    .keyBy(Transaction::getAccountId)           // shuffle less data
    .sum("amount");

What should a senior engineer know?

A senior engineer should know that every redistribution is the dominant cost in a streaming job, that forwarding chains are essentially free, that changing parallelism between operators forces a redistribution even without keyBy, and that the Web UI shows the job graph with shuffle boundaries highlighted so you can verify where shuffles occur.

JAVA

// Senior insight: changing parallelism forces redistribution
stream.map(txn -> txn.normalize())
    .setParallelism(4)  // parallelism 4
    .map(txn -> txn.enrich())
    .setParallelism(8); // parallelism 8 --> REDISTRIBUTION forced
// Even though both are maps, different parallelism breaks chaining

// Fix: keep same parallelism for chainable operators
stream.map(txn -> txn.normalize())
    .setParallelism(4)
    .map(txn -> txn.enrich())
    .setParallelism(4); // same parallelism --> forwarding + chaining

Visual Overview

DIAGRAM

  Forwarding vs Redistributing

  Forwarding (1:1, no shuffle):
  ┌──────────┐     ┌──────────┐     ┌──────────┐
  │Source-0  │────▶│Filter-0  │────▶│  Map-0   │  (one thread)
  │Source-1  │────▶│Filter-1  │────▶│  Map-1   │  (one thread)
  │Source-2  │────▶│Filter-2  │────▶│  Map-2   │  (one thread)
  │Source-3  │────▶│Filter-3  │────▶│  Map-3   │  (one thread)
  └──────────┘     └──────────┘     └──────────┘
      Each subtask feeds exactly one downstream subtask


  Redistributing (keyBy shuffle):
  ┌──────────┐         ┌───────────┐
  │  Map-0   │──ACC1──▶│  Sum-0    │ (handles ACC-001)
  │          │──ACC3──▶│           │ (handles ACC-003)
  ├──────────┤    ╲    ├───────────┤
  │  Map-1   │──ACC2──▶│  Sum-1    │ (handles ACC-002)
  │          │──ACC1──▶│           │ (handles ACC-004)
  ├──────────┤    ╱    ├───────────┤
  │  Map-2   │──ACC3──▶│  Sum-2    │ (handles ACC-005)
  │          │──ACC2──▶│           │
  └──────────┘         └───────────┘
      Records shuffled by key hash to correct subtask


  Pipeline with Both:
  Source ──▶ Filter ──▶ Map ──║── keyBy ──║──▶ Sum ──▶ Sink
  ←── forwarding chain ──→   shuffle    ←── chain ──→
       (zero cost)           boundary      (zero cost)
                          (expensive)

Key Takeaway

In Flink, every connection between two operators is either a forwarding edge or a redistributing edge, and knowing the difference is central to writing fast jobs. A forwarding edge is a one-to-one connection: when two consecutive operators have the same parallelism and no repartitioning, each upstream subtask feeds exactly one downstream subtask, so records stay in the same partition, order is preserved, and Flink can fuse them into a single task chain running in one thread with no serialization.

A redistributing edge appears whenever records must be repartitioned. The classic cause is keyBy, which hashes each record's key and routes it to the correct subtask. Redistribution is expensive because each record is serialized, sent over the network, and deserialized. Other redistributing operations are rebalance for round-robin load spreading, rescale for local redistribution, and broadcast which sends every record to every subtask.

At the National Bank of India this showed up clearly in our transaction pipeline. The cleaning part, where we dropped test transactions and projected only needed fields, stayed as a forwarding chain fused with the source, costing nothing. Then we needed a per-account running total, which required keyBy on account ID. I deliberately placed filtering and projection before the keyBy so the shuffle carried less data, and I monitored per-subtask load because a few high-volume corporate accounts could skew the hash.

The performance rule is simple: minimize the number of shuffle boundaries, keep as much of the pipeline forwarding as possible, and push filtering before keyBy.

Remember one sentence: Forwarding chains are free because they fuse into one thread, while every keyBy redistribution is a costly shuffle boundary that dominates job performance.

---