文章目录1. 依赖2. 代码1. 依赖dependencies!-- 对于java版本的flink 使用 --dependencygroupIdorg.apache.flink/groupIdartifactIdflink-java/artifactIdversion${flink.version}/version!-- scopeprovided/scope--/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-streaming-java_2.12/artifactIdversion${flink.version}/version!-- scopeprovided/scope--/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-clients_2.12/artifactIdversion${flink.version}/version/dependency!-- 针对于scala语言使用的 --dependencygroupIdorg.apache.flink/groupIdartifactIdflink-scala_2.12/artifactIdversion${flink.version}/versionscopeprovided/scope/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-streaming-scala_2.12/artifactIdversion${flink.version}/versionscopeprovided/scope/dependency/dependencies2. 代码接收来自socketnc -l 9001的数据每两秒进行一次wc聚合//envStreamExecutionEnvironmentenvStreamExecutionEnvironment.getExecutionEnvironment();//source from socketDataStreamSourceStringtextenv.socketTextStream(localhost,9001);//transSingleOutputStreamOperatorTuple2String,IntegerwordCounttext.flatMap(newFlatMapFunctionString,String(){publicvoidflatMap(Stringline,CollectorStringout)throwsException{String[]wordsline.split( );for(Stringword:words){out.collect(word);}}}).map(newMapFunctionString,Tuple2String,Integer(){publicTuple2String,Integermap(Stringword)throwsException{returnnewTuple2String,Integer(word,1);}}).keyBy(newKeySelectorTuple2String,Integer,String(){publicStringgetKey(Tuple2String,Integertup)throwsException{returntup.f0;}}).timeWindow(Time.seconds(2)//两秒的窗口).sum(1);//使用一个线程执行打印操作wordCount.print().setParallelism(1);//执行程序env.execute(SocketWindowWordCountJava);flink datastream程序有几个组成env创建flink job的运行环境source从哪里获取到数据trans对数据具体做哪些处理sink数据写到什么地方execute触发任务的运行
