@@ -0,0 +1,3241 @@
+package fi.aalto.dmg;%0A%0Aimport org.apache.flink.api.common.functions.FlatMapFunction;%0Aimport org.apache.flink.api.common.functions.JoinFunction;%0Aimport org.apache.flink.api.common.typeinfo.IntegerTypeInfo;%0Aimport org.apache.flink.api.java.functions.KeySelector;%0Aimport org.apache.flink.api.java.tuple.Tuple3;%0Aimport org.apache.flink.streaming.api.datastream.DataStream;%0Aimport org.apache.flink.streaming.api.datastream.JoinedStreams;%0Aimport org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;%0Aimport org.apache.flink.streaming.api.windowing.assigners.TumblingTimeWindows;%0Aimport org.apache.flink.streaming.api.windowing.time.Time;%0Aimport org.apache.flink.util.Collector;%0Aimport scala.Tuple2;%0A%0Aimport java.io.Serializable;%0Aimport java.util.concurrent.TimeUnit;%0A%0A/**%0A * Created by jun on 20/11/15.%0A */%0Apublic class JoinTest implements Serializable%7B%0A    private static final long serialVersionUID = -4881298165222782534L;%0A%0A    public static void main(String%5B%5D args) throws Exception %7B%0A        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();%0A%0A        DataStream%3CTuple2%3CString, Integer%3E%3E age = env%0A                .socketTextStream(%22localhost%22, 9999)%0A                .flatMap(new Splitter());%0A%0A        DataStream%3CTuple2%3CString, String%3E%3E sex = env%0A                .socketTextStream(%22localhost%22, 9998)%0A                .flatMap(new Splitter2());%0A%0A        age.join(sex)%0A            .where(new KeySelector%3CTuple2%3CString,Integer%3E, String%3E() %7B%0A                @Override%0A                public String getKey(Tuple2%3CString, Integer%3E value) throws Exception %7B%0A                    return value._1();%0A                %7D%0A            %7D)%0A            .equalTo(new KeySelector%3CTuple2%3CString, String%3E, String%3E() %7B%0A                @Override%0A                public String getKey(Tuple2%3CString, String%3E value) throws Exception %7B%0A                    return value._1();%0A                %7D%0A            %7D)%0A            .window(TumblingTimeWindows.of(Time.of(60, TimeUnit.SECONDS)))%0A                .apply(new JoinFunction%3CTuple2%3CString,Integer%3E, Tuple2%3CString,String%3E, Tuple3%3CString, String, Integer%3E%3E() %7B%0A                    @Override%0A                    public Tuple3%3CString, String, Integer%3E join(Tuple2%3CString, Integer%3E age, Tuple2%3CString, String%3E sex) throws Exception %7B%0A                        return new Tuple3%3CString, String, Integer%3E(sex._1(), sex._2(), age._2());%0A                    %7D%0A                %7D).print();%0A%0A        env.execute(%22Socket Stream WordCount%22);%0A    %7D%0A%0A    public static class Splitter implements FlatMapFunction%3CString, Tuple2%3CString, Integer%3E%3E %7B%0A        @Override%0A        public void flatMap(String sentence, Collector%3CTuple2%3CString, Integer%3E%3E out) throws Exception %7B%0A            for (String name: sentence.split(%22 %22)) %7B%0A                out.collect(new Tuple2%3CString, Integer%3E(name, 1));%0A            %7D%0A        %7D%0A    %7D%0A%0A    public static class Splitter2 implements FlatMapFunction%3CString, Tuple2%3CString, String%3E%3E %7B%0A        @Override%0A        public void flatMap(String sentence, Collector%3CTuple2%3CString, String%3E%3E out) throws Exception %7B%0A            for (String name: sentence.split(%22 %22)) %7B%0A                out.collect(new Tuple2%3CString, String%3E(name, %22M%22));%0A            %7D%0A        %7D%0A    %7D%0A%7D%0A

Add flink join test