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 09:55:23 GMT
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