浅谈实时计算框架Flink集群搭建与运行机制

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-java</artifactId>
        <version>1.7.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.11</artifactId>
        <version>1.7.0</version>
    </dependency>
</dependencies>

public class WordCount {
    public static void main(String[] args) throws Exception {
        // 读取文件数据
        readFile () ;
    }

    public static void readFile () throws Exception {
        // 1、执行环境创建
        ExecutionEnvironment environment = ExecutionEnvironment.getExecutionEnvironment();

        // 2、读取数据文件
        String filePath = "/var/flink/test/word.txt" ;
        DataSet<String> inputFile = environment.readTextFile(filePath);

        // 3、分组并求和
        DataSet<Tuple2<String, Integer>> wordDataSet = inputFile.flatMap(new WordFlatMapFunction(
        )).groupBy(0).sum(1);

        // 4、打印处理结果
        wordDataSet.print();
    }

    // 数据读取个切割方式
    static class WordFlatMapFunction implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String input, Collector<Tuple2<String, Integer>> collector){
            String[] wordArr = input.split(",");
            for (String word : wordArr) {
                collector.collect(new Tuple2<>(word, 1));
            }
        }
    }
}

public class WordCount {
    public static void main(String[] args) throws Exception {
        // 读取端口数据
        readPort ();
    }

    public static void readPort () throws Exception {
        // 1、执行环境创建
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        // 2、读取Socket数据端口
        DataStreamSource<String> inputStream = environment.socketTextStream("hop01", 5566);

        // 3、数据读取个切割方式
        SingleOutputStreamOperator<Tuple2<String, Integer>> resultDataStream = inputStream.flatMap(
                new FlatMapFunction<String, Tuple2<String, Integer>>()
        {
            @Override
            public void flatMap(String input, Collector<Tuple2<String, Integer>> collector) {
                String[] wordArr = input.split(",");
                for (String word : wordArr) {
                    collector.collect(new Tuple2<>(word, 1));
                }
            }
        }).keyBy(0).sum(1);

        // 4、打印分析结果
        resultDataStream.print();

        // 5、环境启动
        environment.execute();
    }
}