大数据（二）flink 2 代码demo

「爱情、让人受尽委屈。」 2023-02-17 02:57 6阅读 0赞

上节是介绍一些概念。下面是具体的例子。

代码传至GITHUB  [传送地址][Link 1]

一般都是flink消费kafka的数据来实时统计数据的

引入flink相关jar包

<properties>
            <flink.version>1.7.0</flink.version>
            <scala.binary.version>2.11</scala.binary.version>
            <kafka-clients>2.5.0</kafka-clients>
        </properties>
        <dependencies>
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-java</artifactId>
                <version>${flink.version}</version>
            </dependency>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
                <version>${flink.version}</version>
            </dependency>
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-connector-filesystem_2.12</artifactId>
                <version>${flink.version}</version>
            </dependency>
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-core</artifactId>
                <version>${flink.version}</version>
            </dependency>
            
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>1.18.12</version>
                <scope>provided</scope>
            </dependency>
            
            <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
                <version>2.5.0</version>
            </dependency>
            
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-connector-kafka-0.11_${scala.binary.version}</artifactId>
                <version>${flink.version}</version>
            </dependency>
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>fastjson</artifactId>
                <version>1.2.50</version>
            </dependency>
            <dependency>
                <groupId>org.apache.hbase</groupId>
                <artifactId>hbase-client</artifactId>
                <version>1.2.12</version>
            </dependency>
        </dependencies>

下面举个最简单的例子，统计传过来的单词的个数。

public class WordCountMain {
    
        public static void main(String[] args) throws Exception {
    
            // 设置flink的运行环境，这里会根据本地环境还是生成环境得到不同的对象。
            final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            //设置并行度1，默认是8
    //        env.setParallelism(1);
    
            //定义加载或创建数据源（source）,监听9000端口的socket消息
            //百度window上安装 netcat ，装完之后命令行 执行 nc -L -p 9000
            DataStream<String> textStream = env.socketTextStream("localhost", 9000, "\n");
    
            //这里的输出类型是String，所以DataStream泛型是String
            //map算子是 1对1的  输入对输出
            //这里将s封装前面加了 (index++) + ".您输入的是："，一般都是转化为对象
    //        DataStream<String> result  = textStream.map(s -> (index++) + ".您输入的是：" + s);
    //        result.print();
    
            //这里的输出类型是String，所以DataStream泛型是String
            DataStream<Tuple2<String, Integer>> result2 = textStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                /**
                 * 这里是将字符串按字符串分组，再统计字符串的数量输出
                 * @param s            输入类型 string
                 * @param collector      输出流   输出类型 Tuple2<String, Integer>， Tuple2是flink定义的自定义类型，当参数有N个时，就用TopleN,目前最多22个
                 * @throws Exception
                 */
                @Override
                public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                    String[] splits = s.toLowerCase().split("\\W+");
                    for (String split : splits) {
                        if (split.length() > 0) {
                            collector.collect(new Tuple2<>(split, 1));
                        }
                    }
                }
            }).returns(Types.TUPLE(Types.STRING, Types.INT)).keyBy(0)           //根据Tuple2的f0字段分组
                    .sum(1);    //根据Tuple2的f1字段累加
            result2.print();
            env.execute("WordCount： ");
        }
    }

测试步骤  将com.shrek.test.flink.wordcount.WordCountMain跑起来，打开命令行 输入各个字母

![20200611162212812.png][]

输出结果

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NocmVrMTE_size_16_color_FFFFFF_t_70][]

这个就是flink的大概计算流程。

一般统计某个实时数据用的，比如当天的交易量，金额，访问量等等这些实时的数据

flink通过source取得数据（可以是kafka,也可以是日志，数据库等），

在通过flink的各个计算因子去map,keyby等操作，

最后将输出结果实时sick到输出段（redis,mysql,hbase等）

其他的应用就将输出结果显示在前端。

[Link 1]: https://github.com/wushu0725/flink-test
[20200611162212812.png]: https://img-blog.csdnimg.cn/20200611162212812.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NocmVrMTE_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200611162412678.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NocmVrMTE=,size_16,color_FFFFFF,t_70