[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]

Re: Flink WindowedStream - Need assistance

Hi Titus,

have you looked into ProcessFunction? ProcessFunction[1] gives you access to the two important streaming primitives "time" and "state".

So in your case you can decide flexibly what you want to put into state and when you want to set and fire a timer (for clean-up) per key.


[1] https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/stream/operators/process_function.html

Am 17.07.18 um 11:39 schrieb Titus Rakkesh:
Friends, any assistance regarding this?

On Mon, Jul 16, 2018 at 3:44 PM, Titus Rakkesh <titus.rakkesh@xxxxxxxxx>

Dear All,

We have 2 independent streams which will receive elements in different

DataStream<Tuple3<String, Integer, Double>> splittedActivationTuple;

DataStream<Tuple2<String, Double>> unionReloadsStream;

We have a requirement to keep "splittedActivationTuple" stream elements in
a Window of eviction time period of 24 hours. So I created a
"WindowedStream" like below,

WindowedStream<Tuple3<String, Integer, Double>, Tuple, GlobalWindow> keyedWindowedActStream = splittedActivationTuple
             .assignTimestampsAndWatermarks(new IngestionTimeExtractor()).keyBy(0).window(GlobalWindows.create())
             .evictor(TimeEvictor.of(Time.of(24, TimeUnit.HOURS)));

Our requirements are following,


    When "unionReloadsStream" receives data, we need to check whether the
    corresponding "String" field matches with the "String" field in the
    WindowedStream and accumulate "WindowedStream's" Double with
    "unionReloadsStream" Double.Will this possible with Flink? I checked
    CoGroup and CoMap. But I couldn't figure out how to do since I am new.

    CEP functionality to create a new Stream of from WindowedStream if the
    Double value > 100? I went through several flink's CEP tutorials. But
    couldn't able to figure out how to do with "WindowedStream"?

I am very new to flink. Any assistance would be highly appreciated.