Skip to main content
hegdemahendra
hegdemahendra
Gold ⭐️⭐️
August 10, 2022
Question

Apache Flink CEP , pattern not matching

Hello Experts,
 

I am new to Flink CEP and trying to test basic things - In below code my expectation is all the input should matched in patter and should print as matched result. But somehow nothing is matching ('matechedStream.print()') any idea about the reason ?

Any suggestion/help would be much appreciated.

 

package com.o9.flink;
import com.o9.flink.asyncio.DemandSupply;
import org.apache.flink.cep.CEP;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;


public class DemandSupplyPattern {

public static void main(String[] args) throws Exception {

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
DataStream<String> keyedInputStream = env.fromElements("AAA","BBB","CCC");

Pattern<String, ?> dspattern = Pattern.<String>begin("start");

PatternStream<String> patternStream = CEP.pattern(keyedInputStream, dspattern);
DataStream<String> matechedStream = patternStream.process(new PatternProcessFunction<String, String>() {
@Override
public void processMatch(Map<String, List<String>> map, Context context, Collector<String> collector) throws Exception {
collector.collect(map.get("start").toString());
}
});

matechedStream.print();

env.execute("DemandSupply-CEP");
}
}


<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-runtime</artifactId>
<version>${flink.version}</version>
<!--<scope>test</scope>-->
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-core</artifactId>
<version>${flink.version}</version>
<!--<scope>provided</scope>-->
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-avro</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java</artifactId>
<version>${flink.version}</version>
<!--<scope>provided</scope>-->
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-statebackend-rocksdb</artifactId>
<version>${flink.version}</version>
<!--<scope>test</scope>-->
</dependency>
<dependency>
<groupId>org.asynchttpclient</groupId>
<artifactId>async-http-client</artifactId>
<version>2.2.0</version>
</dependency>

<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-cep</artifactId>
<version>${flink.version}</version>
<!--<scope>provided</scope>-->
</dependency>


</dependencies>

1 reply

araujo
Platinum ⭐️⭐️⭐️
araujo
Platinum ⭐️⭐️⭐️
August 28, 2022

@hegdemahendra ,

 

You need to set the time characteristic of the stream for it to work. For example, try setting it to processing time, as shown below:

		DataStream<String> matechedStream =  patternStream
				.inProcessingTime()
				.process(new PatternProcessFunction<String, String>() {
					@Override
					public void processMatch(Map<String, List<String>> map, Context context, Collector<String> collector) throws Exception {
						collector.collect(map.get("start").toString());
					}
				}
		);

 

Cheers,

André

Translate
-- Was your question answered? Please take some time to click on "Accept as Solution" below this post. If you find a reply useful, say thanks by clicking on the thumbs up button.
Terms & ConditionsAccessibility statement