Can't apply pattern by key in KeyedStream with Flink CEP

Question

I'm trying to emmit an alert for each 2 occurences of events with same ID. To do this, I'm using keyBy() to apply my operations according to the ID.

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.{TimeCharacteristic, scala}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.api.scala._
import org.apache.flink.cep.CEP
import org.apache.flink.cep.pattern.Pattern
import org.apache.flink.core.fs.FileSystem
import org.apache.flink.streaming.api.datastream.{DataStream, KeyedStream}
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time

class Pipeline {

    //{...}

    var env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)

    //{...}

    var stream : DataStream[String] = env.readTextFile("/home/luca/Desktop/input").name("Stream original")

    var tupleStream = stream.map(new S2TMapFunction())
    var newTupleStream : DataStream[(String,Double,Double,String,Int,Int)] = tupleStream.assignTimestampsAndWatermarks(new PlacasPunctualTimestampAssigner())
    newTupleStream = newTupleStream.process(new RemoveLateDataProcessFunction).keyBy(new TupleKeySelector())

    Pattern.begin[(String,Double,Double,String,Int,Int)]("teste").where(new EventoTesteConditionFunction(2)).times(1)

    val patternStream = CEP.pattern(newTupleStream,pattern)

    val result = patternStream.process(new EventoTestePatternProcessFunction())

    result.writeAsText("/home/luca/Desktop/output",FileSystem.WriteMode.OVERWRITE)

    env.execute()

}

Here is my TupleKeySelector:

import org.apache.flink.api.java.functions.KeySelector

class TupleKeySelector() extends KeySelector[(String,Double,Double,String,Int,Int),Int]{

    override def getKey(value: (String, Double, Double, String, Int, Int)): Int = {

        value._6

    }
}

Here's my PlacasPunctualTimestampAssigner:

import java.sql.Timestamp
import java.text.SimpleDateFormat

import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
import org.apache.flink.streaming.api.watermark.Watermark

class PlacasPunctualTimestampAssigner extends AssignerWithPunctuatedWatermarks[(String,Double,Double,String,Int,Int)]{

    var counter : Int = 0
    var eventsUntilNextWatermark : Int = 0
    var lastTimestamp : Long = _

    override def checkAndGetNextWatermark(lastElement: (String, Double, Double, String, Int, Int), extractedTimestamp: Long): Watermark = {

        if(counter == eventsUntilNextWatermark){

            counter = 0

            var time = new Timestamp(lastTimestamp)
            println("Watermark: ",time.toString)
            new Watermark(lastTimestamp)

        }else{

            null

        }

    }

    override def extractTimestamp(element: (String,Double, Double,String,Int,Int), previousElementTimestamp: Long): Long = {


        val formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")
        val date = formatter.parse(element._4)
        val timestampAtual = new Timestamp(date.getTime).getTime
        lastTimestamp = Math.max(lastTimestamp,timestampAtual)

        //counter = counter + 1

        timestampAtual

    }
}

Here is my EventoTesteConditionFunction:

import org.apache.flink.cep.pattern.conditions.IterativeCondition

class EventoTesteConditionFunction(var counter : Int) extends IterativeCondition[(String,Double,Double,String,Int,Int)] {

    private var c  : Int = 0

    override def filter(value: (String, Double, Double, String, Int, Int), ctx: IterativeCondition.Context[(String, Double, Double, String, Int, Int)]): Boolean = {

        if(c == counter-1){

            println("Event: "+value._6.toString)

            c = 0

            true

        }else{

            c = c + 1

            false

        }

    }
}

According to the semantics adopted by Flink, I should achieve my goal by using times(1) quantifier with my pattern, or even ommitting it. Nevertheless, that's not what I obtained in my output. Here's my test case:

//Input events:

(name,timestamp,id) 
(a,2013-02-08 00:14:00,1)
(b,2013-02-08 00:15:00,1) // Event should be emitted here
(c,2013-02-08 00:15:10,1)
(d,2013-02-08 00:17:15,1) // Event should be emitted here
(e,2013-02-08 00:18:20,1)
(f,2013-02-08 00:19:30,0)
(g,2013-02-08 00:20:35.507,2)
(h,2013-02-08 00:21:39.906,2) // Event should be emitted here
(i,2013-02-08 00:22:03.499,2)
(j,2013-02-08 00:23:23.818,2) // Event should be emitted here
(k,2013-02-08 00:24:02.319,2)
(l,2013-02-08 00:25:30.971,0)

//Expected output(I'm returning only the id value):

1
1
2
2

//Real output:

1
1
0
2
2
0

This test shows that the pattern is not being applied by key, and is instead seeing all the events as belonging to the same group, which is the behaviour of a non-keyed DataStream.

If I set the quantifier to times(2), that's what I get from the same input:

//Expected output:

1
2

//Real output:

0
0

I'm also receiving strange prints from my EventoTesteConditionFunction. It seems like some events are being checked more than once and others are being skipped.

//println from accepted events ofEventoTesteConditionFunction for times(2) quantifier

Event: (b,1)
Event: (c,1)
Event: (d,1)
Event: (e,1)
Event: (f,0)
Event: (h,2)
Event: (i,2)
Event: (j,2)
Event: (k,2)
Event: (l,0)
Event: (l,0)

Given all this, my questions are:

How can I apply the pattern by keys?
Why is Flink presenting such a strange behaviour?

Dawid Wysakowicz Dawid Wysakowicz · Accepted Answer · 2020-03-03T10:01:40

The pattern that you are using:

Pattern.begin[(String,Double,Double,String,Int,Int)]("teste").where(new EventoTesteConditionFunction(2)).times(1)

Says that you want to emit every record that matches your EventoTesteConditionFunction. For your use case you should rather use a pattern:

Pattern.begin[(String,Double,Double,String,Int,Int)]("teste").where(new AlwaysTrue()).times(2)

where AlwaysTrue returns true for all records.

The reason why you are seeing weird behaviour is because how you use the variable c in EventoTesteConditionFunction. This variable is not scoped to the key. Flink will use a single instance of EventoTesteConditionFunction for a single Task, which handles a subgroup of keys rather than a single key. Nevertheless results of the computation are scoped to the key. Internal state determined by the result of the function is also scoped to the key.

On a different note. For such a use case I would rather recommend looking into ProcessFunction. In the documentation you can also see a similar example what you are trying to do that, but using a Flink's state, that will be checkpointed and also scoped to a key.

Can't apply pattern by key in KeyedStream with Flink CEP

1 Answers