Flink 1.17教程：DataStream实现Wordcount——读socket（无界流）

心已赠人 2023-10-15 14:36 86阅读 0赞

## pom.xml ##

<properties>
            <flink.version>1.17.0</flink.version>
        </properties>
     
        <dependencies>
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-streaming-java</artifactId>
                <version>${flink.version}</version>
            </dependency>
     
            <dependency>
                <groupId>org.apache.flink</groupId>
                <artifactId>flink-clients</artifactId>
                <version>${flink.version}</version>
            </dependency>
        </dependencies>

## 代码 ##

### 流处理实现WordCount\_无界流 ###

> 读取 socket 文本流
> 
> 在实际的生产环境中，真正的数据流其实是无界的，有开始却没有结束，这就要求我们需要持续地处理捕获的数据。为了模拟这种场景，可以监听 socket 端口，然后向该端口不断地发送数据。

[atguigu@node001 ~]$ sudo yum install -y netcat
    
    [atguigu@node001 ~]$ nc -lk 7777

![img][]

### DataStream实现Wordcount：读socket（无界流） ###

package com.atguigu.wc;
     
    import org.apache.flink.api.common.typeinfo.TypeHint;
    import org.apache.flink.api.common.typeinfo.TypeInformation;
    import org.apache.flink.api.common.typeinfo.Types;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.configuration.Configuration;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.util.Collector;
     
    /**
     * TODO DataStream实现Wordcount：读socket（无界流）
     *
     */
    public class WordCountStreamUnboundedDemo {
        
        public static void main(String[] args) throws Exception {
        
            // TODO 1. 创建执行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            // IDEA运行时，也可以看到webui，一般用于本地测试
            // 需要引入一个依赖 flink-runtime-web
            // 在idea运行，不指定并行度，默认就是 电脑的 线程数
            // StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
     
            env.setParallelism(3);
     
            // TODO 2. 读取数据： socket
            DataStreamSource<String> socketDS = env.socketTextStream("node001", 7777);
     
            // TODO 3. 处理数据: 切换、转换、分组、聚合
            SingleOutputStreamOperator<Tuple2<String, Integer>> sum = socketDS
                    .flatMap(
                            (String value, Collector<Tuple2<String, Integer>> out) -> {
        
                                String[] words = value.split(" ");
                                for (String word : words) {
        
                                    out.collect(Tuple2.of(word, 1));
                                }
                            }
                    )
                    .setParallelism(2)
                    .returns(Types.TUPLE(Types.STRING, Types.INT))
                    // .returns(new TypeHint<Tuple2<String, Integer>>() {})
                    .keyBy(value -> value.f0)
                    .sum(1);
     
            // TODO 4. 输出
            sum.print();
     
            // TODO 5. 执行
            env.execute();
        }
    }
     
    /**
     * 并行度的优先级：
     * 代码：算子 > 代码：env > 提交时指定 > 配置文件
     */

### 演示、对比 ###

![img][img 1]

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

links：

[https://blog.csdn.net/weixin\_44949135/article/details/130895033][https_blog.csdn.net_weixin_44949135_article_details_130895033]

[img]: https://img-blog.csdnimg.cn/ec70b5a8a1074f27b5808faedb9abd12.png
[img 1]: https://img-blog.csdnimg.cn/692f702b4193431d9a9adf231789d9ad.png
[https_blog.csdn.net_weixin_44949135_article_details_130895033]: https://blog.csdn.net/weixin_44949135/article/details/130895033