CEP doesn't seem to be supported by pyflink. Is there a plan to add it? Interestingly ChatGPT misleads that it is available with the following code snippet. Not sure where it comes from. 😅

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream import TimeCharacteristic
from pyflink.datastream.pattern import Pattern
from pyflink.datastream.functions import KeyedProcessFunction

# create a StreamExecutionEnvironment
env = StreamExecutionEnvironment.get_execution_environment()
env.set_stream_time_characteristic(TimeCharacteristic.EventTime)

# create a DataStream of events
input_stream = env.from_elements(
    (1, "start", 1.0),
    (2, "start", 2.0),
    (3, "end", 3.0),
    (4, "start", 4.0),
    (5, "end", 5.0)
).assign_timestamps_and_watermarks(WatermarkStrategy.for_monotonous_timestamps())

# define a pattern to match the start and end events
pattern = Pattern.begin("start").where(lambda event: event[1] == "start").followed_by("end").where(lambda event: event[1] == "end")

# define a function to process the matched events
class MyProcessFunction(KeyedProcessFunction):
    def process_element(self, event, ctx, collector):
        print(f"matched event: {event}")

# apply the pattern on the input stream
matched_events = pattern.match(input_stream).process(MyProcessFunction())

# execute the dataflow
matched_events.execute("CEP example")

Why not use MATCH_RECOGNIZE in PyFlink?

@Martijn Visser +1. IMO, ChatGPT doesn't know the function of. CEP in Flink SQL. @Jaehyeon Kim you could use the table API of PyFlink to use the

MATCH_RECOGNIZE

.