Apache Flink

https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/fault-tolerance/broadcast_state/
Admin
Apache Flink is a distributed stream processing framework for real-time data processing and stateful...
Tokens:2,504
Snippets:16
Trust Score:10
Update:6 months ago
Show doc for...
Tokens:
Raw
### Implement KeyedBroadcastProcessFunction Example

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/fault-tolerance/broadcast_state

An example implementation of a KeyedBroadcastProcessFunction showing how to define MapStateDescriptors for managing state within a keyed stream context.

```java
new KeyedBroadcastProcessFunction<Color, Item, Rule, String>() {
    private final MapStateDescriptor<String, List<Item>> mapStateDesc =
        new MapStateDescriptor<>(
            "items",
            BasicTypeInfo.STRING_TYPE_INFO,
            new ListTypeInfo<>(Item.class));

    private final MapStateDescriptor<String, Rule> ruleStateDescriptor = 
        new MapStateDescriptor<>(
            "RulesBroadcastState",
            BasicTypeInfo.STRING_TYPE_INFO,
            BasicTypeInfo.of(Rule.class));
};
```

--------------------------------

### Keying Item Stream by Color - Java

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/fault-tolerance/broadcast_state

This snippet demonstrates how to key a stream of 'Item' objects by their 'Color' property. This ensures that all items of the same color are processed on the same physical machine, which is a prerequisite for matching items of the same color.

```java
KeyedStream<Item, Color> colorPartitionedStream = itemStream
                        .keyBy(new KeySelector<Item, Color>(){
    @Override
    public Color getKey() throws Exception {
        // Implementation to extract Color as key
        return null; // Placeholder
    }
});
```

--------------------------------

### Connecting Keyed and Broadcast Streams for Processing - Java

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/fault-tolerance/broadcast_state

This snippet illustrates how to connect a keyed stream ('colorPartitionedStream') with a broadcast stream ('ruleBroadcastStream'). It uses a 'KeyedBroadcastProcessFunction' to define the logic for processing items against the broadcasted rules, enabling pattern detection based on color and evolving rules.

```java
DataStream<String> output = colorPartitionedStream
                 .connect(ruleBroadcastStream)
                 .process(
                     
                     // type arguments in our KeyedBroadcastProcessFunction represent: 
                     //   1. the key of the keyed stream
                     //   2. the type of elements in the non-broadcast side
                     //   3. the type of elements in the broadcast side
                     //   4. the type of the result, here a string
                     
                     new KeyedBroadcastProcessFunction<Color, Item, Rule, String>() {
                         // my matching logic
                     }
                 );
```

--------------------------------

### Define BroadcastProcessFunction and KeyedBroadcastProcessFunction

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/fault-tolerance/broadcast_state

Abstract class definitions for BroadcastProcessFunction and KeyedBroadcastProcessFunction, showing the required process methods for handling broadcasted and non-broadcasted streams.

```java
public abstract class BroadcastProcessFunction<IN1, IN2, OUT> extends BaseBroadcastProcessFunction {
    public abstract void processElement(IN1 value, ReadOnlyContext ctx, Collector<OUT> out) throws Exception;
    public abstract void processBroadcastElement(IN2 value, Context ctx, Collector<OUT> out) throws Exception;
}

public abstract class KeyedBroadcastProcessFunction<KS, IN1, IN2, OUT> {
    public abstract void processElement(IN1 value, ReadOnlyContext ctx, Collector<OUT> out) throws Exception;
    public abstract void processBroadcastElement(IN2 value, Context ctx, Collector<OUT> out) throws Exception;
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<OUT> out) throws Exception;
}
```

--------------------------------

### Broadcasting Rules and Creating Broadcast State - Java

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/fault-tolerance/broadcast_state

This code snippet shows how to broadcast a stream of 'Rule' objects to all downstream tasks. It utilizes a 'MapStateDescriptor' to define how the rules will be stored in the broadcast state, allowing each task to have a local, up-to-date set of rules.

```java
// a map descriptor to store the name of the rule (string) and the rule itself.
MapStateDescriptor<String, Rule> ruleStateDescriptor = new MapStateDescriptor<>(
			"RulesBroadcastState",
			BasicTypeInfo.STRING_TYPE_INFO,
			TypeInformation.of(new TypeHint<Rule>() {}));
		
// broadcast the rules and create the broadcast state
BroadcastStream<Rule> ruleBroadcastStream = ruleStream
                        .broadcast(ruleStateDescriptor);
```

--------------------------------

### Flink BroadcastState Process Functions in Java

Source: https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/fault-tolerance/broadcast_state

Implements a Flink BroadcastProcessFunction to manage broadcast state. It handles processing broadcast elements to update the state and processing regular elements by querying the broadcast state for matches and performing stateful logic. Dependencies include Flink's DataStream API, specifically BroadcastProcessFunction, MapState, and TypeInformation.

```java
public class RuleProcessor extends BroadcastProcessFunction<Item, Rule, String> {
    private MapStateDescriptor<String, List<Item>> mapStateDesc;
    private ValueStateDescriptor<Rule> ruleStateDescriptor;

    public RuleProcessor(MapStateDescriptor<String, List<Item>> mapStateDesc, ValueStateDescriptor<Rule> ruleStateDescriptor) {
        this.mapStateDesc = mapStateDesc;
        this.ruleStateDescriptor = ruleStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // Initialize state descriptors if not passed in constructor, or for other states
        // Example for MapState:
        // mapStateDesc = new MapStateDescriptor<>("rule-items", String.class, TypeInformation.of(new TypeHint<List<Item>>() {}));
        // Example for ValueState (if needed for broadcast side):
        // ruleStateDescriptor = new ValueStateDescriptor<>("rule-state", TypeInformation.of(Rule.class));
    }

    @Override
    public void processBroadcastElement(Rule value,
                                        Context ctx,
                                        Collector<String> out) throws Exception {
        ctx.getBroadcastState(ruleStateDescriptor).put(value.name, value);
    }

    @Override
    public void processElement(Item value,
                               ReadOnlyContext ctx,
                               Collector<String> out) throws Exception {

        final MapState<String, List<Item>> state = getRuntimeContext().getMapState(mapStateDesc);
        final Shape shape = value.getShape();
    
        for (Map.Entry<String, Rule> entry : 
                ctx.getBroadcastState(ruleStateDescriptor).immutableEntries()) {
            final String ruleName = entry.getKey();
            final Rule rule = entry.getValue();
    
            List<Item> stored = state.get(ruleName);
            if (stored == null) {
                stored = new ArrayList<>();
            }
    
            if (shape == rule.second && !stored.isEmpty()) {
                for (Item i : stored) {
                    out.collect("MATCH: " + i + " - " + value);
                }
                stored.clear();
            }
    
            // there is no else{} to cover if rule.first == rule.second
            if (shape.equals(rule.first)) {
                stored.add(value);
            }
    
            if (stored.isEmpty()) {
                state.remove(ruleName);
            } else {
                state.put(ruleName, stored);
            }
        }
    }
}
```
Tokens:
Raw