Adding new rules at run time in apache kafka flink

Question

I am using FlinkKafka for applying rules over the stream. And the following is the sample code:

ObjectMapper mapper = new ObjectMapper();
List<JsonNode> rulesList = null;
try {
    // Read rule file
    rulesList = mapper.readValue(new File("ruleFile"), new TypeReference<List<JsonNode>>(){});

} catch (IOException e1) {
    System.out.println( "Error reading Rules file.");
    System.exit(-1);
}


for (JsonNode jsonObject : rulesList) {
    String id = (String) jsonObject.get("Id1").textValue();

    // Form the pattern dynamically
    Pattern<JsonNode, ?> pattern = null;
    pattern = Pattern.<JsonNode>begin("start").where(new SimpleConditionImpl(jsonObject.get("rule1")));
    // Create the pattern stream
    PatternStream<JsonNode> patternStream = CEP.pattern(data, pattern);

}

But the problem is, FlinkKafka only reads the file once when we start the program and I want the new rules to be added dynamically at runtime and applied to the stream.

Is there any way we can achieve this in Flink Kafka?

David Anderson David Anderson · Accepted Answer · 2018-12-26T17:10:03

Flink's CEP library doesn't (yet) support dynamic patterns. (See FLINK-7129.)

The standard approach for this is to use broadcast state to communicate and store the rules throughout the cluster, but you'll have to come up with some way to evaluate/execute the rules.

See https://training.da-platform.com/exercises/taxiQuery.html and https://github.com/dataArtisans/flink-training-exercises/blob/master/src/main/java/com/dataartisans/flinktraining/examples/datastream_java/broadcast/BroadcastState.java for examples.

Adding new rules at run time in apache kafka flink

1 Answers