【wordcount】flink 代码快速入门

1. 依赖

<dependencies><!-- 对于java版本的flink 使用 --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>${flink.version}</version><!-- <scope>provided</scope>--></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java_2.12</artifactId><version>${flink.version}</version><!-- <scope>provided</scope>--></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients_2.12</artifactId><version>${flink.version}</version></dependency><!-- 针对于scala语言使用的 --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-scala_2.12</artifactId><version>${flink.version}</version><scope>provided</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-scala_2.12</artifactId><version>${flink.version}</version><scope>provided</scope></dependency></dependencies>

2. 代码

• 接收来自socket（nc -l 9001）的数据
• 每两秒进行一次wc聚合

//envStreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnvironment();//source from socketDataStreamSource<String>text=env.socketTextStream("localhost",9001);//trans：SingleOutputStreamOperator<Tuple2<String,Integer>>wordCount=text.flatMap(newFlatMapFunction<String,String>(){publicvoidflatMap(Stringline,Collector<String>out)throwsException{String[]words=line.split(" ");for(Stringword:words){out.collect(word);}}}).map(newMapFunction<String,Tuple2<String,Integer>>(){publicTuple2<String,Integer>map(Stringword)throwsException{returnnewTuple2<String,Integer>(word,1);}}).keyBy(newKeySelector<Tuple2<String,Integer>,String>(){publicStringgetKey(Tuple2<String,Integer>tup)throwsException{returntup.f0;}}).timeWindow(Time.seconds(2)//两秒的窗口).sum(1);//使用一个线程执行打印操作wordCount.print().setParallelism(1);//执行程序env.execute("SocketWindowWordCountJava");

flink datastream程序有几个组成：

1. env：创建flink job的运行环境
2. source：从哪里获取到数据
3. trans：对数据具体做哪些处理
4. sink：数据写到什么地方
5. execute：触发任务的运行