Flink操练（五）之DS简介（5）编写第一个Flink程序

　　Scala版Flink程序编写package learn  import org.apache.flink.api.java.utils.ParameterTool import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment} import org.apache.flink.streaming.api.windowing.time.Time  /**  * @program: bigdata_learn  * @description: ${description}  * @author: Mr.逗  * @create: 2021-09-08 10:04  **/ object StreamingJob {   def main(args: Array[String]): Unit = {     //获取socket端口号     val port:Int = try{       ParameterTool.fromArgs(args).getInt(＂port＂)     }catch {       case e: Exception=>{         System.err.println(＂No port set. Use default port 9000!＂)       }         9000     }     //获取运行环境     val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment     //链接socket获取输入数据     val text =  env.socketTextStream(＂172.17.0.50＂,9000,＂ ＂)      import org.apache.flink.api.scala._     //解析数据(数据扁平化)，分组，窗口计算，并且聚合求sum     val wordWithCount = text.flatMap(line => line.split(＂,＂)).map(w =>WordWithCount(w,1))       .keyBy(＂word＂)//分组       .timeWindow(Time.seconds(2),Time.seconds(1))//指定窗口大小，间隔时间       //.sum(＂count＂);//sum 或者reduce 都行       .reduce((a,b)=>WordWithCount(a.word,a.count+b.count));      wordWithCount.print().setParallelism(1); //打印到控制台      env.execute(＂Socket Window Count!＂)   }   case class WordWithCount(word:String,count:Long) }
　　打开一个终端（Terminal），运行以下命令 $ nc -lk 9999
　　接下来使用 IDEA  运行就可以了。Java版Flink程序编写import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.util.Collector;  public class WordCountFromSocket {      public static void main(String[] args) throws Exception {         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();         env.setParallelism(1);          DataStream stream = env.socketTextStream(＂localhost＂, 9999);          stream.flatMap(new Tokenizer()).keyBy(r -> r.f0).sum(1).print();          env.execute(＂Flink Streaming Java API Skeleton＂);     }      public static class Tokenizer implements FlatMapFunction> {         @Override         public void flatMap(String value, Collector> out) throws Exception {             String[] stringList = value.split(＂s＂);             for (String s : stringList) {                 // 使用out.collect方法向下游发送数据                 out.collect(new Tuple2(s, 1));             }         }     } }2 下载Flink运行时环境，提交Jar包的运行方式
　　下载链接：http://mirror.bit.edu.cn/apache/flink/flink-1.11.1/flink-1.11.1-bin-scala_2.11.tgz
　　然后解压 $ tar xvfz flink-1.11.1-bin-scala_2.11.tgz
　　启动Flink集群 $ cd flink-1.11.1 $ ./bin/start-cluster.sh
　　可以打开Flink WebUI查看集群状态：http://localhost:8081
　　在 IDEA  中使用maven package  打包。
　　提交打包好的 JAR  包$ cd flink-1.11.1 $ ./bin/flink run 打包好的JAR包的绝对路径
　　停止Flink集群 $ ./bin/stop-cluster.sh
　　查看标准输出日志的位置，在 log  文件夹中。$ cd flink-1.11.1/log