Flink 学习笔记 2. 流处理WordCount

Love The Way You Lie 2022-12-29 14:21 211阅读 0赞

> Flink version: 1.11.3

**目录**

1. StreamWordCount.java

2. 执行效果

3. 关于并行度

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

#  1. StreamWordCount.java #

这里使用 socketTextStream 作为流式数据源

package com.mit.wc;
    
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    /**
     * @Author mit
     * @Description 流处理WordCount
     * @Date 2020/12/23 上午10:59
     * @Version 1.0
     */
    public class StreamWordCount {
        public static void main(String[] args) throws Exception{
            // 创建执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            // 设置并行度, 并行度默认为CPU核数
            env.setParallelism(1);
    
            // 监听本地socket端口获取数据流
            DataStream<String> inputDataStream = env.socketTextStream("127.0.0.1", 8888);
    
            // 对数据流进行转换计算
            DataStream<Tuple2<String, Integer>> resultStream = inputDataStream.flatMap(new WordCount.MyFlatMapper())
                    .keyBy(value -> value.f0)
                    .sum(1);
    
            resultStream.print();
    
            // 执行task
            env.execute();
        }
    }

# 2. 执行效果 #

使用nc往8888端口发送数据，查看task执行效果。(注意，需要先运行nc命令)

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70][]

# 3. 关于并行度 #

task执行的默认并行度为CPU核数，上面设置了并行度为1 (**env.setParallelism(1);**)，注释并行度设置代码行，观察输出结果

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70 1][]

可以看到每个输出的二元组前面都有 **数字+>,** 这里的数字代表的是并发的线程编号，我的机器为8核CPU，所以最大为8；

# 4. 从启动参数获取配置信息 #

上面是将监听的地址端口直接硬编码在程序中，实际使用中应将地址端口配置使用

调整代码，修改为从启动参数获取配置信息

package com.mit.wc;
    
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.api.java.utils.ParameterTool;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    /**
     * @Author mit
     * @Description 流处理WordCount
     * @Date 2020/12/23 上午10:59
     * @Version 1.0
     */
    public class StreamWordCount {
        public static void main(String[] args) throws Exception{
            // 创建执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            // 设置并行度, 并行度默认为CPU核数
            // env.setParallelism(1);
    
            // 从启动参数获取配置信息
            ParameterTool parameterTool = ParameterTool.fromArgs(args);
            String host = parameterTool.get("host", "localhost");
            Integer port = parameterTool.getInt("port", 8888);
    
            // 监听本地socket端口获取数据流
            DataStream<String> inputDataStream = env.socketTextStream(host, port);
    
            // 对数据流进行转换计算
            DataStream<Tuple2<String, Integer>> resultStream = inputDataStream.flatMap(new WordCount.MyFlatMapper())
                    .keyBy(value -> value.f0)
                    .sum(1);
    
            resultStream.print();
    
            // 执行task
            env.execute();
        }
    }

在运行配置中加入启动参数  **--host localhost --port 8888**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70 2][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70 3][]

运行效果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70 4][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70]: /images/20221120/20867858aae646ab8b680875f9457ac4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70 1]: /images/20221120/fc964e4309db4878ab9cb1fdd83dcfc9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70 2]: /images/20221120/d61766185c9446cdb6715c8ce9b858d7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70 3]: /images/20221120/d70c8027ef04491f9efeb66cfebaa9ed.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MDIzODYyNQ_size_16_color_FFFFFF_t_70 4]: /images/20221120/29bd3010ab0641439962bc1de72b6f03.png