Mailing-List: contact dev-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@flink.apache.org
MIME-Version: 1.0
In-Reply-To: <26F1352E-3493-4AEA-BFF6-F7A99924E73C@alibaba-inc.com>
References: <CAMpYU5QdE5g8m8MbZWg6iVwBVVoCzw1De4vyrJVWkGwt+sC8ZA@mail.gmail.com>
 <575F1989.8070003@apache.org> <CAMpYU5QF_9aBRmbVMGPbFsU+L=6DNVjw1BDkTVFO2njCh3H-bA@mail.gmail.com>
 <26F1352E-3493-4AEA-BFF6-F7A99924E73C@alibaba-inc.com>
From: Vinay Patil <vinay18.patil@gmail.com>
Date: Tue, 14 Jun 2016 13:37:18 +0530
Message-ID: <CAMpYU5R5ig=NGR79t+p2zwjdaPTM_GhCCuToH7sZ0gsY6CDNBw@mail.gmail.com>
Subject: Re: [Discussion] Query regarding Join
To: dev@flink.apache.org
Content-Type: multipart/alternative; boundary=001a114457f66db8ce0535387e12
archived-at: Tue, 14 Jun 2016 08:07:34 -0000

--001a114457f66db8ce0535387e12
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: quoted-printable

You are right, debugged it for all elements , I can do that now.
Thanks a lot.

Regards,
Vinay Patil

On Tue, Jun 14, 2016 at 11:56 AM, Jark Wu <wuchong.wc@alibaba-inc.com>
wrote:

> In `coGroup(Iterable<Integer> iter1, Iterable<Integer> iter2,
> Collector<Integer> out)` ,   when both iter1 and iter2 are not empty, it
> means they are matched elements from both stream.
> When one of iter1 and iter2 is empty , it means that they are unmatched.
>
>
> - Jark Wu (wuchong)
>
> > =E5=9C=A8 2016=E5=B9=B46=E6=9C=8814=E6=97=A5=EF=BC=8C=E4=B8=8B=E5=8D=88=
12:46=EF=BC=8CVinay Patil <vinay18.patil@gmail.com> =E5=86=99=E9=81=93=EF=
=BC=9A
> >
> > Hi Matthias ,
> >
> > I did not get you, even if we use Co-Group we have to apply it on a key
> >
> > sourceStream.coGroup(destStream)
> > .where(new ElementSelector())
> > .equalTo(new ElementSelector())
> > .window(TumblingEventTimeWindows.of(Time.seconds(30)))
> > .apply(new CoGroupFunction<Integer, Integer, Integer>() {
> > private static final long serialVersionUID =3D 6408179761497497475L;
> >
> > @Override
> > public void coGroup(Iterable<Integer> paramIterable, Iterable<Integer>
> > paramIterable1,
> > Collector<Integer> paramCollector) throws Exception {
> > Iterator<Integer> iterator =3D paramIterable.iterator();
> > while(iterator.hasNext()) {
> > }
> > }
> > });
> >
> > when I debug this ,only the matched element from both stream will come =
in
> > the coGroup function.
> >
> > What I want is how do I check for unmatched elements from both streams
> and
> > write it to sink.
> >
> > Regards,
> > Vinay Patil
> >
> > *+91-800-728-4749*
> >
> > On Tue, Jun 14, 2016 at 2:07 AM, Matthias J. Sax <mjsax@apache.org>
> wrote:
> >
> >> You need to do an outer-join. However, there is no build-in support fo=
r
> >> outer-joins yet.
> >>
> >> You can use Window-CoGroup to implement the outer-join as an own
> operator.
> >>
> >>
> >> -Matthias
> >>
> >> On 06/13/2016 06:53 PM, Vinay Patil wrote:
> >>> Hi,
> >>>
> >>> I have a question regarding the join operation, consider the followin=
g
> >>> dummy example:
> >>>
> >>> StreamExecutionEnvironment env =3D
> >>> StreamExecutionEnvironment.getExecutionEnvironment();
> >>> env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
> >>> DataStreamSource<Integer> sourceStream =3D
> >>> env.fromElements(10,20,23,25,30,33,102,18);
> >>> DataStreamSource<Integer> destStream =3D
> >> env.fromElements(20,30,40,50,60,10);
> >>>
> >>> sourceStream.join(destStream)
> >>> .where(new ElementSelector())
> >>> .equalTo(new ElementSelector())
> >>> .window(TumblingEventTimeWindows.of(Time.milliseconds(10)))
> >>> .apply(new JoinFunction<Integer, Integer, Integer>() {
> >>>
> >>> private static final long serialVersionUID =3D 1L;
> >>>
> >>> @Override
> >>> public Integer join(Integer paramIN1, Integer paramIN2) throws
> Exception
> >> {
> >>> return paramIN1;
> >>> }
> >>> }).print();
> >>>
> >>> I perfectly get the elements that are matching in both the streams,
> >> however
> >>> my requirement is to write these matched elements and also the
> unmatched
> >>> elements to sink(S3)
> >>>
> >>> How do I get the unmatched elements from each stream ?
> >>>
> >>> Regards,
> >>> Vinay Patil
> >>>
> >>
> >>
>
>

--001a114457f66db8ce0535387e12--