美文网首页flink简单使用教程
flink使用02-从WordCount开始

flink使用02-从WordCount开始

作者: CheckChe | 来源:发表于2019-11-09 21:52 被阅读0次

    相信大家在学习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));
                    }
                }
            }
        }
    }
    

    相关文章

      网友评论

        本文标题:flink使用02-从WordCount开始

        本文链接:https://www.haomeiwen.com/subject/plxobctx.html