【零基础学flink】flink实战(idea版本)

川长思鸟来 2022-02-14 07:47 246阅读 0赞

### 回顾 ###

上一小节中我们介绍了flink的安装，以及flink word count的实现，本文主要介绍如何使用idea开发flink项目

### 一、新建Maven项目 ###

这一步比较简单，网上也有很多教程，主要是使用maven来管理依赖包

### 二、pom的配置 ###

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

### 三、wordcount实战（在线版本） ###

import org.apache.flink.api.common.functions.FlatMapFunction;
    import org.apache.flink.api.java.utils.ParameterTool;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.windowing.time.Time;
    import org.apache.flink.util.Collector;
    
    /** * Author: Mr.Deng * Date: 2018/10/15 * Desc: 使用flink对指定窗口内的数据进行实时统计，最终把结果打印出来 * 先在node21机器上执行nc -l 9000 */
    public class StreamingWindowWordCountJava { 
        public static void main(String[] args) throws Exception { 
            //定义socket的端口号
            int port;
            try{ 
                ParameterTool parameterTool = ParameterTool.fromArgs(args);
                port = parameterTool.getInt("port");
            }catch (Exception e){ 
                System.err.println("没有指定port参数，使用默认值6100");
                port = 6100;
            }
            //获取运行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //连接socket获取输入的数据
            DataStreamSource<String> text = env.socketTextStream("211.1xx.1xx.6x", port, "\n");
    // 本文填写的是远程linux ip，在远程linux上需要执行：nc -l 6100命令
            //计算数据
            DataStream<WordWithCount> windowCount = text.flatMap(new FlatMapFunction<String, WordWithCount>() { 
                public void flatMap(String value, Collector<WordWithCount> out) throws Exception { 
                    String[] splits = value.split("\\s");
                    for (String word:splits) { 
                        out.collect(new WordWithCount(word,1L));
                    }
                }
            })//打平操作，把每行的单词转为<word,count>类型的数据
                    //针对相同的word数据进行分组
                    .keyBy("word")
                    //指定计算数据的窗口大小和滑动窗口大小
                    .timeWindow(Time.seconds(5),Time.seconds(1))
                    .sum("count");
            //把数据打印到控制台,使用一个并行度
            windowCount.print().setParallelism(1);
            //注意：因为flink是懒加载的，所以必须调用execute方法，上面的代码才会执行
            env.execute("streaming word count");
        }
    
        /** * 主要为了存储单词以及单词出现的次数 */
        public static class WordWithCount{ 
            public String word;
            public long count;
            public WordWithCount(){ }
            public WordWithCount(String word, long count) { 
                this.word = word;
                this.count = count;
            }
    
            @Override
            public String toString() { 
                return "WordWithCount{" +
                        "word='" + word + '\'' +
                        ", count=" + count +
                        '}';
            }
        }
    
    }

### 四、离线版本wordCount ###

import org.apache.flink.api.common.functions.FlatMapFunction;
    import org.apache.flink.api.java.DataSet;
    import org.apache.flink.api.java.ExecutionEnvironment;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.util.Collector;
    
    /** /** * Author: Mr.Deng * Date: 2018/10/19 * Desc: */
    public class WordCountJava { 
        public static void main(String[] args) throws Exception { 
            //构建环境
            final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
            //通过字符串构建数据集
            DataSet<String> text = env.fromElements(
                    " Who is there ? ",
                    " I think I hear them. Stand, ho ! Who is there ? ");
            //分割字符串、按照key进行分组、统计相同的key个数
            DataSet<Tuple2<String, Integer>> wordCounts = text
                    .flatMap(new LineSplitter())
                    .groupBy(0)
                    .sum(1);
            //打印
            wordCounts.print();
        }
        //分割字符串的方法
        public static class LineSplitter implements FlatMapFunction<String, Tuple2<String, Integer>> { 
            public void flatMap(String line, Collector<Tuple2<String, Integer>> out) { 
                for (String word : line.split(" ")) { 
                    out.collect(new Tuple2<String, Integer>(word, 1));
                }
            }
        }
    }

--------------------

扫描下方二维码，**及时**获取更多**互联网求职面经**、**java**、**python**、**爬虫**、**大数据**等技术，和**海量资料分享**：  
公众号\*\*`菜鸟名企梦`**后台发送“csdn”即可免费领取【csdn】和【百度文库】下载服务；  
公众号**`菜鸟名企梦`**后台发送“资料”:即可领取**5T精品学习资料\*\*、**java面试考点**和**java面经总结**，以及**几十个java、大数据项目**，**资料很全，你想找的几乎都有**  
![扫码关注，及时获取更多精彩内容。（博主今日头条大数据工程师）][1240]

[1240]: /images/20220214/10438c783f6a44889a015b3bf9403524.png