重庆分公司,新征程启航
为企业提供网站建设、域名注册、服务器等服务
这篇文章主要介绍“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怎么用”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注创新互联网站,小编会继续努力为大家带来更多实用的文章!