使用RichFunction实现累计字符串长度

### 使用 `RichFunction` 实现累计字符串长度

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

如下代码示例是一个使用 Apache Flink 编写的流处理应用，其主要功能是从自定义数据源 `NBUSource` 中实时接收字符串流，并通过使用`RichFunction`来实现累加器的功能，计算自程序启动以来接收到的所有字符串的总长度。

```java
package org.nbubigdata.flink;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.configuration.Configuration;

public class AccumulatorDemo {
    public static void main(String[] args) throws Exception {
        // 初始化Flink流执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

//env.setParallelism(1);

// 添加自定义数据源，产生字符串流
        env.addSource(new NBUSource())
                // 使用RichMapFunction对流中的每个字符串进行处理
                .map(new RichMapFunction<String, Integer>() {
                    private transient int totalLength; // 用于累计字符串长度的变量

// 在函数实例化时初始化累加器
                    @Override
                    public void open(Configuration parameters) {
                        totalLength = 0;
                    }

// 对于每个字符串，将其长度加到累加器上，并输出当前累加器的值
                    @Override
                    public Integer map(String value) {
                        totalLength += value.length();
                        return totalLength; // 返回累计的字符串长度
                    }
                })
                // 将累加器的当前值输出
                .print("Accumulated String Length");

// 执行Flink程序
        env.execute("Accumulator with RichFunction for String Length");
    }
}
```

运行上述程序，可以在控制台中观察到如下的输出信息。由于运行程序的机器的CPU核心数为2，Flink作业会利用多个核心进行并行处理任务。这个过程展示了 Flink 的能力在处理实时数据流时进行简单的数据转换和聚合操作。

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

大数据学习

导航

最近发表

友情链接