flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Ashwin Jayaprakash <ashwin.jayaprak...@gmail.com>
Subject Self-join with filter
Date Mon, 17 Aug 2015 13:30:48 GMT
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