Flink案例分析

public class MapTest {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet dataSet = env.fromElements(1, 2, -3, 0, 5, -9, 8);
        DataSet dataSet2 = dataSet.map(new Tokenizer());
//        DataSet dataSet2 = dataSet.map(i->i * 2);
        dataSet2.print();
    }

    public static class Tokenizer implements MapFunction {
        @Override
        public Integer map(Integer in) {
            return in * 2;
        }
    }
}

public class SocketWindowWordCountJava {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet dataSet = env.readTextFile("/yourpath/in.txt");

        DataSet> counts =
                // split up the lines in pairs (2-tuples) containing: (word,1)
                dataSet.flatMap(new Tokenizer())
                        // group by the tuple field "0" and sum up tuple field "1"
                        .groupBy(0)
                        .sum(1);

        String outputPath = "/yourpath/out.txt";
        counts.writeAsCsv(outputPath, "\n", " ");
        env.execute("myflink");
    }

    public static class Tokenizer implements FlatMapFunction> {
        @Override
        public void flatMap(String value, Collector> out) {
            String[] tokens = value.split(" ");
            // emit the pairs
            for (String token : tokens) {
                if (token.length() > 0) {
                    out.collect(new Tuple2(token, 1));
                }
            }
        }
    }
}

public class ReduceTest {
    private static final Logger LOG = LoggerFactory.getLogger(ReduceTest.class);
    private static final String[] TYPE = {"苹果", "梨", "西瓜", "葡萄", "火龙果"};

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //添加自定义数据源,每秒发出一笔订单信息{商品名称,商品数量}
        DataStreamSource> orderSource = env.addSource(new SourceFunction>() {
            private volatile boolean isRunning = true;
            private final Random random = new Random();

            @Override
            public void run(SourceContext> ctx) throws Exception {
                while (isRunning) {
                    TimeUnit.SECONDS.sleep(1);
                    ctx.collect(Tuple2.of(TYPE[random.nextInt(TYPE.length)], 1));
                }
            }

            @Override
            public void cancel() {
                isRunning = false;
            }

        }, "order-info");

        orderSource.keyBy(0)
                //将上一元素与当前元素相加后，返回给下一元素处理
                .reduce(new ReduceFunction>() {
                    @Override
                    public Tuple2 reduce(Tuple2 value1, Tuple2 value2)
                            throws Exception {
                        return Tuple2.of(value1.f0, value1.f1 + value2.f1);
                    }
                })
                .print();

        env.execute("Flink Streaming Java API Skeleton");
    }
}