网站建设资讯

NEWS

网站建设资讯

FlinkProcess怎么用

这篇文章主要介绍“Flink  Process怎么用”,在日常操作中,相信很多人在Flink  Process怎么用问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”Flink  Process怎么用”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!

创新互联公司是一家企业级云计算解决方案提供商,超15年IDC数据中心运营经验。主营GPU显卡服务器,站群服务器,服务器托管,海外高防服务器,大带宽服务器,动态拨号VPS,海外云手机,海外云服务器,海外服务器租用托管等。

process算子:处理每个keyBy(分区)输入到窗口的批量数据流(为KeyedStream类型数据流)

示例环境

java.version: 1.8.x
flink.version: 1.11.1

示例数据源 (项目码云下载)

Flink 系例 之 搭建开发环境与数据

Process.java

import com.flink.examples.DataSource;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;
import java.util.Iterator;
import java.util.List;
/**
 * @Description process算子:处理每个keyBy(分区)输入到窗口的批量数据流(为KeyedStream类型数据流)
 */
public class Process {

    /**
     * 遍历集合,分别打印不同性别的总人数与年龄之和
     * @param args
     * @throws Exception
     */
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        List> tuple3List = DataSource.getTuple3ToList();
        DataStream dataStream = env.fromCollection(tuple3List)
                .keyBy((KeySelector, String>) k -> k.f1)
                //按数量窗口滚动,每3个输入数据流,计算一次
                .countWindow(3)
                //处理每keyBy后的窗口数据流,process方法通常应用于KeyedStream类型的数据流处理
                .process(new ProcessWindowFunction, String, String, GlobalWindow>() {
                    /**
                     * 处理窗口数据集合
                     * @param s         从keyBy里返回的key值
                     * @param context   窗口的上下文
                     * @param input     从窗口获取的所有分区数据流
                     * @param out       输出数据流对象
                     * @throws Exception
                     */
                    @Override
                    public void process(String s, Context context, Iterable> input, Collector out) throws Exception {
                        Iterator> iterator = input.iterator();
                        int total = 0;
                        int i = 0;
                        while (iterator.hasNext()){
                            Tuple3 tuple3 = iterator.next();
                            total += tuple3.f2;
                            i ++ ;
                        }
                        out.collect(s + "共:"+i+"人,平均年龄:" + total/i);
                    }
                });
        dataStream.print();
        env.execute("flink Process job");
    }
}

打印结果

4> girl共:3人,平均年龄:24
2> man共:3人,平均年龄:26

到此,关于“Flink  Process怎么用”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注创新互联网站,小编会继续努力为大家带来更多实用的文章!


分享名称:FlinkProcess怎么用
标题来源:http://cdweb.net/article/pjjhph.html