相信大家在学习spark的时候接触的第一个案例肯定也是 wordCount, 本文也想通过这样一个简单的例子来讲一下一个简单的 flink 程序是什么样子的, 让大家对 flink 的代码有一个简单的了解.

一个 flink程序主要分为5个部分:

• 1. 获取执行 Environment
  environment 提供控制 job 执行的方法(例如设置并行度/容错/checkpoint 参数) 并且与外部系统做交互. flink可以做流计算也可以做批计算, 对应的也就有不同的environment , 在使用时根据不同的使用场景选择即可.

• 2. 获取输入流 Source

  一个流式计算框架自然是少不了数据的输入, 在 streamExecutionEnvironment 的可以看到有很多种创建输入流的方式, 不过在项目中使用最多的还是使用 addSource()方法来添加不同的数据源接入

  数据源接入方法.png

• 3. 执行计算 Operator

  在spark中,对数据的转换计算是通过 action 算子和 transformation 算子来对 RDD 中的数据来进行操作, 而在flink中, 主要是通过 Operator来对一个流做处理, 通过一个 Operator 可以将一个流转换为另外一个流, flink中内置了很多算子来实现Operator操作.

• 4. 输入结果 Sink

  在完成数据计算之后,就需要有一个输出的地方, 通常来讲也是通过 addSink() 方法来添加不同的数据输出目标,也可以通过 print() 来直接查看输出或者写入到csv等外部文件.

• 5. 启动 flink,提交 job

  一个 flink 代码的启动执行, 必须通过 env.executor() 方法.这行代码主要做了以下事情:

  1. 生成StreamGraph
  2. 生成JobGraph.
  3. 生成一系列配置
  4. 将 JobGraph和配置交给 flink 集群去运行
  5. 以本地模式运行的话,可以看到启动过程,如启动能量度,web模块,jobManager,ResourceManager,taskManager等等
  6. 启动任务

以下为简单的 WordCount 代码

public class WordCount {

    public static void main(String[] args) throws Exception {

        // 获取 StreamEnv
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 获取 输入流
        DataStream<String> text = env.fromElements(WordCountData.WORDS);

        // 执行计算Operator
        DataStream<Tuple2<String, Integer>> counts
                = text.flatMap(new SplitFunction())
                .keyBy(0).sum(1);

        // 输出结果
        counts.print();

        // 启动flink程序
        env.execute("WordCount Demo");
    }

    // *************************************************************************
    // 自定义切割Function切分一行输入
    // *************************************************************************
    public static final class SplitFunction implements FlatMapFunction<String, Tuple2<String, Integer>>{

        @Override
        public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
            String[] words = s.toLowerCase().split(" ");
            for (String word : words) {
                if (word.length() > 0){
                    collector.collect(new Tuple2<>(word, 1));
                }
            }
        }
    }
}