flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Titus Rakkesh <titus.rakk...@gmail.com>
Subject Re: Window Stream - Need assistance
Date Wed, 18 Jul 2018 13:44:11 GMT
Thanks Xingcan. I specified as GlobalWindow since I am going to put all the
elements coming with splittedActivationTuple with a 24 hour expiry and then
do operations on that when elements coming with stream "unionReloadsStream"
(bigger set).

On Wed, Jul 18, 2018 at 4:07 PM, Xingcan Cui <xingcanc@gmail.com> wrote:

> Hi Rakkesh,
>
> The `GlobalWindow` is commonly used for custom window assignment and you
> should specify a `trigger` for it [1].
> If the built-in window (e.g., tumbling window or sliding window) join in
> DataStream API fails to meet the requirements,
> you could try the time-windowed join in Table/SQL API [2].
>
> Hope that helps.
>
> Best,
> Xingcan
>
> [1] https://ci.apache.org/projects/flink/flink-docs-
> master/dev/stream/operators/windows.html#global-windows
> [2] https://ci.apache.org/projects/flink/flink-docs-
> master/dev/table/sql.html#joins
>
>
> On Jul 18, 2018, at 5:55 PM, Titus Rakkesh <titus.rakkesh@gmail.com>
> wrote:
>
> Thanks for the reply. I have called "env.execute()". But nothing getting
> printed. I have a doubt whether "implemented function" is correct with my
> "requirement". Please assist.
>
> On Wed, Jul 18, 2018 at 2:57 PM, Xingcan Cui <xingcanc@gmail.com> wrote:
>
>> Hi Rakkesh,
>>
>> Did you call `execute()`on your `StreamExecutionEnvironment`?
>>
>> Best,
>> Xingcan
>>
>> > On Jul 18, 2018, at 5:12 PM, Titus Rakkesh <titus.rakkesh@gmail.com>
>> wrote:
>> >
>> > Dear Friends,
>> >          I have 2 streams of the below data types.
>> >
>> > DataStream<Tuple3<String, Integer, Double>> splittedActivationTuple;
>> >
>> > DataStream<Tuple2<String, Double>> unionReloadsStream;
>> >
>> > These streams are getting data from Kafka and getting data in different
>> frequencies. "unionReloadsStream"  will receive more data than
>> "splittedActivationTuple". I need to store  "splittedActivationTuple" in a
>> Window of 24 hours and manipulate its "Double" field, if a matching data
>> comes from unionReloadsStream (String field is the common field).
>> >
>> > So I wrote the following method to do this task.
>> >
>> >
>> > public static DataStream<Tuple3<String, Integer, Double>>
>> joinActivationsBasedOnReload(
>> >             DataStream<Tuple3<String, Integer, Double>>
>> activationsStream,
>> >             DataStream<Tuple2<String, Double>> unifiedReloadStream)
{
>> >
>> >         return activationsStream.join(unifiedReloadStream).where(new
>> ActivationStreamSelector())
>> >                 .equalTo(new ReloadStreamSelector()).window
>> (GlobalWindows.create())
>> >                 .evictor(TimeEvictor.of(Time.of(24, TimeUnit.HOURS)))
>> >                 .apply(new JoinFunction<Tuple3<String, Integer,
>> Double>, Tuple2<String, Double>, Tuple3<String, Integer, Double>>()
{
>> >                     private static final long serialVersionUID = 1L;
>> >                     @Override
>> >                     public Tuple3<String, Integer, Double>
>> join(Tuple3<String, Integer, Double> first,
>> >                             Tuple2<String, Double> second) {
>> >                         return new Tuple3<String, Integer,
>> Double>(first.f0, first.f1, first.f2 + second.f1);
>> >                     }
>> >                 });
>> >     }
>> >
>> >
>> > and calling as,
>> >
>> > DataStream<Tuple3<String, Integer, Double>> activationWindowStream
=
>> joinActivationsBasedOnReload(splittedActivationTuple,
>> unionReloadsStream);
>> >
>> > activationWindowStream.print();
>> >
>> >
>> > But I couldn't see anything printing.
>> >
>> > I expected "activationWindowStream" to contain the
>> "splittedActivationTuple" (smaller set) data and the Double value
>> accumulated if  unionReloadsStream's incoming elements have a matching
>> "String" field. But that is not happening. Where I am missing?
>> >
>> > Thanks,
>> > Rakkesh
>>
>>
>
>

Mime
View raw message