使用keyBy对无界数据流进行分区

### 使用keyBy对无界数据流进行分区

> Written by 黄俊仁，2024/04/13

如下示例实现了一个动态单词计数功能。程序从自定义数据源 `NBUSource` 接收一个连续的字符串流，然后通过实现的 `Tokenizer` 类将每个字符串拆分成单词，并为每个单词生成一个包含单词本身和初始计数 1 的元组。这些元组随后根据单词（元组的第一个元素）进行分组，对每个组内的计数（元组的第二个元素）进行累加，以此计算出每个单词在流中出现的总次数。

```java
package org.nbubigdata.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class StreamKeyBy {
    public static void main(String[] args) throws Exception {
        // 设置流执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 使用自定义的数据源NBUSource
        DataStream<String> text = env.addSource(new NBUSource());

// 将输入字符串转换为(word, 1)形式的元组，并按word字段进行分组
        DataStream<Tuple2<String, Integer>> counts = text
                .flatMap(new Tokenizer()) // 将每行文本拆分为单词并初始化计数值
                .keyBy(value -> value.f0) // 按Tuple的第一个字段分组
                .sum(1); // 对每个分组的第二个元素进行求和

// 打印结果
        counts.print();

// 执行程序
        env.execute("Streaming Word Count");
    }

// 自定义FlatMapFunction，用于拆分输入的字符串，并为每个单词生成(word, 1)形式的元组
    public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
            // 按空格拆分输入的字符串
            for (String word : value.split("\\s")) {
                // 为每个单词输出一个(word, 1)的元组
                out.collect(new Tuple2<>(word, 1));
            }
        }
    }
}
```

运行上述程序，可以在控制台中观察到如下的输出信息。`keyBy` 会根据提供的键选择器函数对流中的每个元素提取键，并根据这个键将元素分配到不同的任务（或者称为任务槽）上。这样，所有具有相同键的元素都会被发送到相同的任务上进行处理，保证了按键处理的一致性和正确性。

![image-20240408221107274](https://hexo-img.obs.cn-east-3.myhuaweicloud.com/img/image-20240408221107274.png)

大数据学习

导航

最近发表

友情链接