flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Ashwin Jayaprakash <ashwin.jayaprak...@gmail.com>
Subject Re: Self-join with filter
Date Wed, 19 Aug 2015 02:55:00 GMT
Stephan, this is exactly what I was looking for :) Thanks, will try it out.

I know the combineGroup() needed a reduceGroup() too, but I was just trying
out the APIs.

I did not realize that the other streaming APIs were already available. I
will have a look.

Thanks again.



On Mon, Aug 17, 2015 at 6:30 AM, Ashwin Jayaprakash <
ashwin.jayaprakash@gmail.com> wrote:

> Hi, I'm trying to evaluate Flink to see if it can do efficient semi-joins
> or self-joins with filter.
>
> Problem description:
> I have 1 stream that can contain "near duplicates" records. The records
> share a "family name" and so, many records can have the same family name.
> But each record has a unique id too.
>
> Query:
> 1) I would like to run a query such that I get the ids and names of some
> records based on a criterion
>
> 2) I then want to fetch all the records that match this set of ids fetched
> previously but this time include the duplicates that share the same name
>
> Question 1:
> I've pasted the sample code below, but I'm trying to see if I can push all
> the ids that came out of the first filter step "into" the connector,
> partition by partition and then pull out only the matching records in step
> 2. Is there a way to do this?
>
> Question 2:
> I haven't seen many examples of non-Hadoop connectors. Is there any plan
> for those?
>
> (If you are curious, I was looking at Presto and had to post the question
> there also -
> https://groups.google.com/forum/#!topic/presto-users/Ns0q4pvHwfo)
>
> public class JoinExample {
>     static final int AGE_ADD_CONSTANT = 100;
>
>     public static void main(String[] args) throws Exception {
>         final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
>
>         DataSet<Record> allRecords = env.fromElements(
>                 new Record("aa", 1), new Record("bb", 2), new Record("ee", 3), new Record("ww",
23),
>                 new Record("cc", 3),
>                 new Record("dd", 1), new Record("bb", 12), new Record("aa", 6), new Record("aa",
33),
>                 new Record("ff", 87));
>         allRecords.print();
>         System.out.println();
>
>         DataSet<Record> newerRecords = allRecords
>                 .filter(new FilterFunction<Record>() {
>                     @Override
>                     public boolean filter(Record user) throws Exception {
>                         return user.age < 10;
>                     }
>                 })
>                 .groupBy(new KeySelector<Record, String>() {
>                     @Override
>                     public String getKey(Record user) throws Exception {
>                         return user.id;
>                     }
>                 })
>                 .combineGroup(new GroupCombineFunction<Record, Record>() {
>                     @Override
>                     public void combine(Iterable<Record> iterable, Collector<Record>
collector) throws Exception {
>                         int age = 0;
>                         String id = null;
>                         for (Record user : iterable) {
>                             if (id == null) {
>                                 id = user.id;
>                             }
>                             //Just some way to fake a merge.
>                             age = (age * AGE_ADD_CONSTANT) + user.age;
>                         }
>                         collector.collect(new Record(id, age));
>                     }
>                 });
>         newerRecords.print();
>         System.out.println();
>
>         DataSet<Record> multiAgeRecords = allRecords
>                 /*
>                 If this were another data set, would it have been possible to push all
newerRecord ids
>                 partition by partition as multiple filters, without having to do an explicit
join?
>                  */
>                 .join(newerRecords)
>                 .where(new KeySelector<Record, String>() {
>                     @Override
>                     public String getKey(Record user) throws Exception {
>                         return user.id;
>                     }
>                 }).equalTo(new KeySelector<Record, String>() {
>                     @Override
>                     public String getKey(Record user) throws Exception {
>                         return user.id;
>                     }
>                 })
>                 .with(new FlatJoinFunction<Record, Record, Record>() {
>                     @Override
>                     public void join(Record user, Record user2, Collector<Record>
collector) throws Exception {
>                         collector.collect(new Record(user.id, user.age));
>                     }
>                 })
>                 .groupBy(new KeySelector<Record, String>() {
>                     @Override
>                     public String getKey(Record user) throws Exception {
>                         return user.id;
>                     }
>                 })
>                 .combineGroup(new GroupCombineFunction<Record, Record>() {
>                     @Override
>                     public void combine(Iterable<Record> iterable, Collector<Record>
collector) throws Exception {
>                         int age = 0;
>                         String id = null;
>                         for (Record user : iterable) {
>                             if (id == null) {
>                                 id = user.id;
>                             }
>                             //Just some way to fake a merge.
>                             age = (age * 100) + user.age;
>                         }
>                         collector.collect(new Record(id, age));
>                     }
>                 });
>
>         multiAgeRecords.print();
>     }
>
>     static class Record {
>         String id;
>         int age;
>
>         Record(String id, int age) {
>             this.id = id;
>             this.age = age;
>         }
>
>         @Override
>         public String toString() {
>             return "{" + "id='" + id + '\'' + ", age=" + age + '}';
>         }
>     }
> }
>
>
>
> Thanks.
>

Mime
View raw message