[
https://issues.apache.org/jira/browse/FLINK-6321?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15998019#comment-15998019
]
Shashank Agarwal commented on FLINK-6321:
-----------------------------------------
Hi [~kkl0u] ,
Actually it's too complicated with kafka streams and custom serializer. Above
steps are correct but still I try tpo put some code. I have modified parameter
names and some things in code. If you find any issue let me know.
{code}
object Job {
def main(args: Array[String]) {
// set up the execution environment
val env = StreamExecutionEnvironment.getExecutionEnvironment
val propertiesFile =
getClass.getClassLoader.getResource("xyz.properties").getPath
val parameter = ParameterTool.fromPropertiesFile(propertiesFile)
env.getConfig.setGlobalJobParameters(parameter)
env.setStateBackend(new RocksDBStateBackend(parameter.get("rocksDBPath")))
// enable fault-tolerance
env.enableCheckpointing(1000)
env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
// enable restarts
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(50, 500L))
val properties = new Properties()
properties.setProperty("bootstrap.servers", parameter.get("kafkaUrl"))
properties.setProperty("group.id", parameter.get("kafkaGroupId"))
val kafka10 = new
FlinkKafkaConsumer010[RawSignal](parameter.get("kafkaBundleName"), new
SignalDeserializationSchema(), properties)
val stream = env.addSource(kafka10).keyBy(_._someKey.getOrElse(0))
//Creating a pattern for successful event
val successOrderPattern = Pattern.begin[RawSignal]("someEvent").
.followedBy("otherEvent")
val successOrderPatternStream = CEP.pattern(stream.keyBy((x) =>
(x._someKey.getOrElse(0), x._someSubKey.getOrElse(0))), successOrderPattern)
val ordersStream: DataStream[TransactionSignal] =
successOrderPatternStream.select(new TransactionPatternFlatMap)
//Put Ip count in the stream with maintaining the state
val ipStateStream = ordersStream.keyBy((x) => (x._someKey, x._deviceIp))
.mapWithState((in: OrderSignal, ipState: Option[Int]) => {
if(!in._deviceIp.equalsIgnoreCase(parameter.get("defaultIp"))) {
val newCount = ipState.getOrElse(0) + 1
val output = in.copy(_numOfOrderSameIp = newCount)
(output, Some(newCount))
}
else
{
(in, Some(0))
}
}
)
ipStateStream.print
env.execute("Thirdwatch Mitra")
{code}
> RocksDB state backend Checkpointing is not working with KeyedCEP.
> -----------------------------------------------------------------
>
> Key: FLINK-6321
> URL: https://issues.apache.org/jira/browse/FLINK-6321
> Project: Flink
> Issue Type: Bug
> Components: CEP
> Affects Versions: 1.2.0
> Environment: yarn-cluster, RocksDB State backend, Checkpointing every
> 1000 ms
> Reporter: Shashank Agarwal
> Assignee: Kostas Kloudas
> Priority: Blocker
> Fix For: 1.3.0
>
>
> Checkpointing is not working with RocksDBStateBackend when using CEP. It's
> working fine with FsStateBackend and MemoryStateBackend. Application failing
> every-time.
> {code}
> 04/18/2017 21:53:20 Job execution switched to status FAILING.
> AsynchronousException{java.lang.Exception: Could not materialize checkpoint
> 46 for operator KeyedCEPPatternOperator -> Map (1/4).}
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask$AsyncCheckpointRunnable.run(StreamTask.java:980)
> at
> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:745)
> Caused by: java.lang.Exception: Could not materialize checkpoint 46 for
> operator KeyedCEPPatternOperator -> Map (1/4).
> ... 6 more
> Caused by: java.util.concurrent.CancellationException
> at java.util.concurrent.FutureTask.report(FutureTask.java:121)
> at java.util.concurrent.FutureTask.get(FutureTask.java:192)
> at
> org.apache.flink.util.FutureUtil.runIfNotDoneAndGet(FutureUtil.java:40)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask$AsyncCheckpointRunnable.run(StreamTask.java:915)
> ... 5 more
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
