2023学年完整公开课版Flink 实时APITransformations实践

Flink实时API-Transformations实践DataStream——TransformationsDataStreamAPISourceTransformationsSinkDataStream——TransformationsDataStreamAPISourceTransformationsSink定义了一系列算子，这些算子能够将一个或多个DataStream转换为新的DataStreamDataStream——

Split

和

Select

算子Split：根据规则把一个数据流切分为多个流Select：和split配合使用，选择切分后的流Steam1Steam2SteamSplitSplitSteam1Steam2Steam1Steam2SplitStreamSplitStreamSelectSelectvards=env.fromElements(("machine","虚拟机启动"),("machine","虚拟机恢复"),("docker","容器启动"),("docker","容器恢复"),("windows","windows启动"),("windows","windows恢复")).split(temp=>{varflaglist:List[String]=nullif("machine".equals(temp._1))flaglist=List("machine")if("docker".equals(temp._1))flaglist=List("docker")if("windows".equals(temp._1))flaglist=List("windows")flaglist})ds.select("docker").print()DataStream——

Split

和

Select

算子DataStream——

Split

和

Select

算子

vards=env.fromElements(("machine","虚拟机启动"),("machine","虚拟机恢复"),("docker","容器启动"),("docker","容器恢复"),("windows","windows启动"),("windows","windows恢复")).split(temp=>{varflaglist:List[String]=nullif("machine".equals(temp._1))flaglist=List("machine")if("docker".equals(temp._1))flaglist=List("docker")if("windows".equals(temp._1))flaglist=List("windows")flaglist})ds.select("docker").print()4>(docker,容器恢复)3>(docker,容器启动)DataStream——

Split

和

Select

算子DatasourceA流B流B-1流A-1流B-2流splitsplitsplitsplitA-2流valenv=StreamExecutionEnvironment.getExecutionEnvironmentvards=env.fromElements(("machine","虚拟机启动"),("machine","虚拟机恢复"),("docker","容器启动"),("docker","容器恢复"),("windows","windows启动"),("windows","windows恢复")

//1先定义各类流的标签

varmachine=newOutputTag[(String,String)]("machine")vardocker=newOutputTag[(String,String)]("docker")varmiddle=newOutputTag[(String,String)]("middle")

DataStream——

Side-outputs算子DataStream——

Side-outputs算子

valds2=cess(newProcessFunction[(String,String),(String,String)]{

overridedefprocessElement(//数据流中的每条数据value:(String,String),//上下文对象，给需要打标签的对象打上标签

ctx:ProcessFunction[(String,String),(String,String)]//不需要打标签的数据，通过out发送出去

out:Collector[(String,String)]):Unit={if("machine".equals(value._1))ctx.output(machine,value)if("docker".equals(value._1))ctx.output(docker,value)if("middle".equals(value._1))ctx.output(middle,value)}})ds2.getSideOutput(machine).print()DataStream——

Side-outputs算子

valds2=cess(newProcessFunction[(String,String),(String,String)]{

overridedefprocessElement(//数据流中的每条数据value:(String,String),//上下文对象，给需要打标签的对象打上标签

ctx:ProcessFunction[(String,String),(String,String)]//不需要打标签的数据，通过out发送出去

out:Collector[(String,String)]):Unit={if("machine".equals(value._1))ctx.output(machine,value)if("docker".equals(value._1))ctx.output(docker,value)if("middle".equals(value._1))ctx.output(middle,value)}})ds2.getSideOutput(machine).print()8>(machine,虚拟机启动)7>(machine,虚拟机恢复)KeyedStream→DataStreamReduce算子，基于keyBy算子，是一个分组数据流的聚合操作，合并当前的元素和上次聚合的结果，产生一个新的值，返回的流中包含每一次聚合的结果，而不是只返回最后一次聚合的最终结果。DataStream——

Reduce算子DataStream——

Reduce算子

//1获取flink执行环境valenv=StreamExecutionEnvironment.getExecutionEnvironment//2创建初始的数据流valds=env.fromElements(("hello",1),("hello",1),("hello",1))//3Reduce算子

valds2=ds.keyBy(0)valds3=ds2.reduce((a,b)=>(a._1,a._2+b._2))//4打印输出到控制台ds3.print()//5触发真正执行env.execute()}}DataStream——

Reduce算子

//1获取flink执行环境valenv=StreamExecutionEnvironment.getExecutionEnvironment//2创建初始的数据流valds=env.fromElements(("hello",1),("he