使用ProcessFunction实现计数器

### 使用 `ProcessFunction` 实现计数器

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

这个程序利用 Apache Flink 的流处理能力，通过自定义数据源 `NBUSource` 生成一个不断变化的字符串流，并使用 `ProcessFunction` 对每个接收到的字符串进行计数。它在处理每个元素时递增计数器，并将当前的计数值输出到控制台。

```java
package org.nbubigdata.flink;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

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

//env.setParallelism(1);

// 添加自定义数据源，产生字符串流
        DataStream<String> dataStream = env.addSource(new NBUSource());

// 使用ProcessFunction对流进行处理
        dataStream.process(new ProcessFunction<String, Integer>() {
                    private transient int count = 0; // 用于计数的变量

// 对每个元素进行处理，每处理一个元素，计数器加一，并输出当前计数值
                    @Override
                    public void processElement(String value, Context ctx, Collector<Integer> out) {
                        count++; // 更新计数器
                        out.collect(count); // 输出当前计数值
                    }
                })
                // 将计数器的当前值输出
                .print("Count");

// 执行Flink程序
        env.execute("Counter with ProcessFunction");
    }
}
```

这个简洁的示例演示了如何在 Flink 中对实时数据流进行基本的处理和状态管理。运行上述程序，可以在控制台中观察到如下的输出信息。由于运行程序的机器的CPU核心数为2，Flink作业会利用多个核心进行并行处理任务。

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

大数据学习

导航

最近发表

友情链接