You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Sachin Mittal <sj...@gmail.com> on 2024/08/07 02:42:24 UTC
Is there any way to perform left outer join using flink's data stream APIs
Hi,
I have two streams A and B.
Which can be joined or connected using a field f.
However, for a given record in A for f = f1, there does not exist any
record in B matching this field f = f1.
In such cases I want to do a left outer join where the combined record
pushed downstream would only have field values from A and empty for ones
supposed to be joined from B.
Please let me know if there is any way I can do this, using the CoProcess
function or something.
Thanks
Sachin
Re: Is there any way to perform left outer join using flink's data stream APIs
Posted by Kirill Ternovsky <ki...@interchecks.com>.
Hard to give any specifics without knowing how your data is being
streamed in. If both sources are in the same datastore, you can always
join at the data source and then emit the joined (left or otherwise)
records to your stream processor. On the other hand, if only Stream A or
B is coming in, you can use a RichAsyncFunction to enrich records one at
a time. On the third hand, if they're both streaming in from different
sources, you can probably correlate them according to processing time,
assuming that you don't expect an enrichment record from stream B to
show up more than N minutes after its corresponding record in Stream A -
in that case you can use a
`TumblingProcessTimeWindow.of(Duration.ofMinutes(N))`.
Best,
Kirill
On 8/6/24 10:24 PM, Sachin Mittal wrote:
> So, I have a question related to windowing:
> None of the records from A or B are timestamped.
>
> The collections are backed by two master tables (each containing say a
> few million records) in a database and I want to use flink to
> basically join them and create an enriched stream C.
>
> Hence I won't be able to use the window operator. Now loading entire
> collections in flink memory may not be feasible, so how best I can do
> this using coGroup.
>
> Thanks
> Sachin
>
>
> On Wed, Aug 7, 2024 at 10:41 AM Kirill Ternovsky
> <ki...@interchecks.com> wrote:
>
> Hi Sachin,
>
> You should be able to use `coGroup`:
>
>
> ```
>
> streamA.coGroup(streamB)
>
> .where(StreamAFieldFSelector).isEqualTo(StreamBFieldFSelector)
>
> .window(/*your windowing spec*/)
>
> .apply(/*your cogroup function*/)
>
> ```
>
>
> Your coGroup function will look something like:
>
> ```
>
> @Override
>
> public void coGroup(Iterable<StreamARecord> arecs,
> Iterable<StreamBRecord> brecs, Collector<TOut> out) {
>
> // if `brecs` is empty, that means nothing from Stream B matched
> Stream A in the window.
>
> }
>
> ```
>
>
> Best,
>
> Kirill
>
> On 8/6/24 7:42 PM, Sachin Mittal wrote:
> > Hi,
> > I have two streams A and B.
> > Which can be joined or connected using a field f.
> >
> > However, for a given record in A for f = f1, there does not
> exist any
> > record in B matching this field f = f1.
> >
> > In such cases I want to do a left outer join where the combined
> record
> > pushed downstream would only have field values from A and empty for
> > ones supposed to be joined from B.
> >
> > Please let me know if there is any way I can do this, using the
> > CoProcess function or something.
> >
> > Thanks
> > Sachin
> >
>
Re: Is there any way to perform left outer join using flink's data stream APIs
Posted by Sachin Mittal <sj...@gmail.com>.
So, I have a question related to windowing:
None of the records from A or B are timestamped.
The collections are backed by two master tables (each containing say a few
million records) in a database and I want to use flink to basically join
them and create an enriched stream C.
Hence I won't be able to use the window operator. Now loading entire
collections in flink memory may not be feasible, so how best I can do this
using coGroup.
Thanks
Sachin
On Wed, Aug 7, 2024 at 10:41 AM Kirill Ternovsky <
kirill.ternovsky@interchecks.com> wrote:
> Hi Sachin,
>
> You should be able to use `coGroup`:
>
>
> ```
>
> streamA.coGroup(streamB)
>
> .where(StreamAFieldFSelector).isEqualTo(StreamBFieldFSelector)
>
> .window(/*your windowing spec*/)
>
> .apply(/*your cogroup function*/)
>
> ```
>
>
> Your coGroup function will look something like:
>
> ```
>
> @Override
>
> public void coGroup(Iterable<StreamARecord> arecs,
> Iterable<StreamBRecord> brecs, Collector<TOut> out) {
>
> // if `brecs` is empty, that means nothing from Stream B matched
> Stream A in the window.
>
> }
>
> ```
>
>
> Best,
>
> Kirill
>
> On 8/6/24 7:42 PM, Sachin Mittal wrote:
> > Hi,
> > I have two streams A and B.
> > Which can be joined or connected using a field f.
> >
> > However, for a given record in A for f = f1, there does not exist any
> > record in B matching this field f = f1.
> >
> > In such cases I want to do a left outer join where the combined record
> > pushed downstream would only have field values from A and empty for
> > ones supposed to be joined from B.
> >
> > Please let me know if there is any way I can do this, using the
> > CoProcess function or something.
> >
> > Thanks
> > Sachin
> >
>
Re: Is there any way to perform left outer join using flink's data stream APIs
Posted by Kirill Ternovsky <ki...@interchecks.com>.
Hi Sachin,
You should be able to use `coGroup`:
```
streamA.coGroup(streamB)
.where(StreamAFieldFSelector).isEqualTo(StreamBFieldFSelector)
.window(/*your windowing spec*/)
.apply(/*your cogroup function*/)
```
Your coGroup function will look something like:
```
@Override
public void coGroup(Iterable<StreamARecord> arecs,
Iterable<StreamBRecord> brecs, Collector<TOut> out) {
// if `brecs` is empty, that means nothing from Stream B matched
Stream A in the window.
}
```
Best,
Kirill
On 8/6/24 7:42 PM, Sachin Mittal wrote:
> Hi,
> I have two streams A and B.
> Which can be joined or connected using a field f.
>
> However, for a given record in A for f = f1, there does not exist any
> record in B matching this field f = f1.
>
> In such cases I want to do a left outer join where the combined record
> pushed downstream would only have field values from A and empty for
> ones supposed to be joined from B.
>
> Please let me know if there is any way I can do this, using the
> CoProcess function or something.
>
> Thanks
> Sachin
>