一、运行环境介绍

Flink执行环境主要分为本地环境和集群环境，本地环境主要为了方便用户编写和调试代码使用，而集群环境则被用于正式环境中，可以借助Hadoop Yarn、k8s 或 Mesos等不同的资源管理器部署自己的应用。
环境依赖：【1】JDK环境：Flink核心模块均使用 Java开发，所以运行环境需要依赖 JDK，JDK版本需要保证在1.8以上。
【2】Maven编译环境：Flink的源代码目前仅支持通过 Maven进行编译，所以如果需要对源代码进行编译，或通过 IDE开发Flink Application，则建议使用 Maven作为项目工程编译方式。需要注意的是，Flink程序需要 Maven的版本在3.0.4及以上，否则项目编译可能会出问题，建议用户根据要求进行环境的搭建。
【3】Hadoop环境：对于执行在Hadoop Yarn资源管理器的 Flink应用，则需要配置对应的 Hadoop环境参数。目前 Flink官方提供的版本支持hadoop2.4、2.6、2.7、2.8等主要版本，所以用户可以在这些版本的Hadoop Yarn中直接运行自己的 Flink应用，而不需要考虑兼容性的问题。

二、Flink项目模板

Flink为了对用户使用 Flink进行应用开发进行简化，提供了相应的项目模板来创建开发项目，用户不需要自己引入相应的依赖库，就能够轻松搭建开发环境，前提是在JDK（1.8及以上）和Maven（3.0.4及以上）的环境已经安装好且能正常执行。在 Flink项目模板中，Flink提供了分别基于Java和Scala实现的模板，下面就 Java项目模板进行介绍和应用。

【1】创建项目：通过 Maven archetype命令进行创建：在项目目录下执行如下命令（第一行就行），注意：groupId、artifactId、version、package 等信息根据自己的需求进行修改，当然你也可以用我默认的。

F:\Project\flink_project>mvn archetype:generate     -DarchetypeGroupId=org.apache.flink     -DarchetypeArtifactId=flink-quickstart-java     -DarchetypeVersion=1.10.0     -DgroupId=my-flink-project     -DartifactId=my-flink-project     -Dversion=0.1     -Dpackage=myflink     -DinteractiveMode=false执行结果如下表示成功
[INFO] ------------------------------------------------------------------------
[INFO] BUILD SUCCESS
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 59.208 s
[INFO] Finished at: 2020-07-01T15:38:00+08:00
[INFO] Final Memory: 18M/312M
[INFO] ------------------------------------------------------------------------

【2】我们可以看到的项目结构如下图所示，可以看到自己定义的 artifactId等信息。该项目已经是一个相对比较完善的 Maven项目，其中创建出来对应的Java实例代码，分别是 BatchJob.java和 Streaming.java两个文件，分别对应 Flink批量接口DataSet的实例代码和流式接口 DataStream的实例代码。在创建好上述项目后，建议用户将项目导入到 IDE进行后续开发。启动IntelliJ IDEA，选择File→Open，在文件选择框中选择创建好的项目（quickstart），点击确定，IDEA将自动进行项目的导入；

【3】编译项目：项目经过上述步骤创建后，可以使用 Maven Command命令 mvn clean package对项目进行编译，编译完成后在项目同级目录会生成 target/-.jar，则该可执行 Jar包就可以通过 Flink命令或者 Web客户端提交到集群上执行。

注意： 通过 Maven创建Java应用，用户可以在 Pom中指定 Main Class，这样提交执行过程中就具有默认的入口Main Class，否则需要用户在执行的 Flink App的Jar应用中指定Main Class。

【4】开发应用：在项目创建和检测完成后，我们就可以选择在模板项目中的代码上编写应用，也可以定义 Class调用DataSet API或DataStream API进行Flink应用的开发，然后通过编译打包，上传并提交到集群上运行。

三、Flink开发环境配置（自动生成可忽略）

对于通过项目模板生成的项目，项目中的主要参数配置已被初始化，所以无须额外进行配置，如果用户通过手工进行项目的创建，则需要创建Flink项目并进行相应的基础配置，包括Maven Dependences、Scala的Version等配置信息。
【1】Flink基础依赖库：对于Java版本，需要在项目的 pom.xml文件中配置如下依赖库，其中 flink-java和 flink-streaming-java分别是批量计算 DataSet API和流式计算 DataStream API的依赖库，{flink.version}是官方的发布的版本号，用户可根据自身需要进行选择：

<flink.version>1.6.1</flink.version>
<dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>${flink.version}</version><scope>compile</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java_${scala.binary.version}</artifactId><version>${flink.version}</version><scope>compile</scope></dependency>
</dependencies>

另外在上述 Maven Dependences配置中，核心的依赖库配置的Scope为provided，主要目的是在编译阶段能够将依赖的Flink基础库排除在项目之外，当用户提交应用到 Flink集群的时候，就避免因为引入Flink基础库而导致 Jar包太大或类冲突等问题。而对于 Scope配置成 provided的项目可能出现本地 IDE中无法运行的问题，可以在 Maven中通过配置 Profile的方式，动态指定编译部署包的 scope为provided，本地运行过程中的 scope为compile，从而解决本地和集群环境编译部署的问题。

scope 的作用范围：
【1】compile：默认值，适用于所有阶段（表明该 jar 包在编译、运行以及测试中路径均可见），并且会随着项目一起发布。
【2】test：只在测试时使用，用于编译和运行测试代码，不会随项目发布。
【3】runtime：无需参与项目的编译，不过后期的测试和运行周期需要其参与，与 compile 相比，跳过了编译。如 JDBC 驱动，适用运行和测试阶段。
【4】provided：编译和测试时有效，但是该依赖在运行时由服务器提供，并且打包时也不会被包含进去。如 servlet-api。
【5】system：类似 provided，需要显式提供包含依赖的jar，不会从 maven 仓库下载，而是从本地文件系统获取，需要添加 systemPath 的属性来定义路径。

【2】 Flink Connector 和 Lib依赖库：除了上述Flink项目中应用开发必须依赖的基础库之外，如果用户需要添加其他依赖，例如Flink中內建的Connector，或者其他第三方依赖库，需要在项目中添加相应的Maven Dependences，并将这些Dependence的Scope需要配置成 compile。如果项目中需要引入Hadoop相关依赖包，和基础库一样，在打包编译的时候将 Scope注明为 provided，因为 Flink集群中已经将 Hadoop依赖包添加在集群的环境中，用户不需要再将相应的 Jar包打入应用中，否则容易造成 Jar包冲突。

注意：对于有些常用的依赖库，为了不必每次都要上传依赖包到集群上，用户可以将依赖的包可以直接上传到 Flink安装部署路径中的 lib目录中，这样在集群启动的时候就能够将依赖库加载到集群的 ClassPath中，无须每次在提交任务的时候上传依赖的 Jar包。

四、编写 Flink 程序

在 src/main/java/myflink 下创建 SocketWindowWordCount.java 文件：具体步骤见注解

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.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;/*** @description: 第一个 flink wordcount 项目* @author: zzx* @createDate: 2020/7/1* @version: 1.0*/
public class SocketWindowWordCount {public static void main(String[] args) {//1、Flink 程序的第一步是创建一个 StreamExecutionEnvironment 流执行环境 。//   这是一个入口类，可以用来设置参数和创建数据源以及提交任务。所以让我们把它添加到 main 函数中：StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//2、将创建一个从本地端口号 9000 的 socket ,通过连接 socket 获取输入数据，如果端口被占，自行更换DataStream<String> text = env.socketTextStream("localhost", 9000, "\n");/*** 这创建了一个字符串类型的 DataStream。* DataStream 是 Flink 中做流处理的核心 API，上面定义了非常多常见的操作（如，过滤、转换、聚合、窗口、关联等）。* 在本示例中，我们感兴趣的是每个单词在特定时间窗口中出现的次数，比如说5秒窗口。* 为此，我们首先要将字符串数据解析成单词和次数（使用Tuple2<String, Integer>表示），* 第一个字段是单词，第二个字段是次数，次数初始值都设置成了1。* 我们实现了一个flatmap，因为一行数据中可能有多个单词。** 解析数据，按 word 分组，开窗，聚合*/SingleOutputStreamOperator<Tuple2<String, Integer>> sum = text.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {@Overridepublic void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {// \\s 表示按照空格拆分for (String word : value.split("\\s")) {out.collect(Tuple2.of(word, 1));}}//根据map中的key进行分组，0表示位置}).keyBy(0).timeWindow(Time.seconds(5))//安装 value 进行统计 1表示位置.sum(1);//将结果打印到控制台,这是设置并行度为1，表示单线程打印，一般我们都是打印到文件的，如果不是1就会打印到多个文件sum.print().setParallelism(1);try {env.execute("Scoket Window WordCount");} catch (Exception e) {e.printStackTrace();}}
}

Linux运行程序：要运行示例程序，首先我们在终端启动 netcat 获得输入流：nc -lk port 永久监听 TCP端口。安装链接

nc -lk 9000

Window运行程序：可以通过 ncat 运行，安装链接

ncat -lk 9000

然后直接运行 SocketWindowWordCount的 main 方法。只需要在 netcat 控制台输入单词，就能在 SocketWindowWordCount 的输出控制台看到每个单词的词频统计。如果想看到大于1的计数，请在5秒内反复键入相同的单词。