??在当前数据激增的时代，各种业务场景都有大量的业务数据。如何有效地处理这些不断生成的数据已经成为大多数公司面临的问题。目前流行的大数据处理引擎Apache Spark，基本上已经取代了MapReduce成为当前大数据处理的标准。但对于实时数据处理，Apache Spark的Spark-Streaming还有提高性能的空间。对于Spark-Streaming本质上，流量计算仍然是批)计算，Apache Flink是近年来开源社区不断发展的纯实时分布式处理框架，可以同时支持高吞吐、低延迟、高性能的技术。

??1) Flink 的发展历史

??柏林工业大学、柏林洪堡大学、哈索普拉特纳研究所于2010年至2014年联合发起名为Stratosphere:Information Management on the Cloud该项目在当时的社区中逐渐具有一定的社区知名度。2014年4月，Stratosphere代码贡献Apache成为软件基金会Apache基金会孵化器项目。参与项目初期的核心成员是Stratosphere一旦核心成员，团队的大多数创始成员离开学校，共同创办了一个名字Data Artisans公司的主要业务是Stratosphere，也就是以后Flink实现商业化。在项目孵化期间，Stratosphere改名为Flink。Flink在德语中，它意味着快速和敏感，用于反映流式数据处理器的快速和灵活性。同时，棕红色松鼠图案被用作Flink项目的Logo，为了突出松鼠灵活快速的特点，Flink正式进入社区开发者的视线。 ??该项目于2014年12月成为Apache2015年9月，软件基金会顶级项目发布了第一个稳定版0.9到目前为止已经发布到1.11的版本，更多的社区开发成员逐步加入，现在Flink全球开发人员350余人，新特色不断发布。与此同时，越来越多的公司开始在全球范围内使用它们Flink，阿里巴巴、美团、滴滴等国内知名互联网公司都在大规模使用Flink作为企业的分布式大数据处理引擎。

??2) Flink的定义

??Apache Flink 用于在无边界和有边界数据流上进行状态计算的框架和分布式处理引擎。Flink 能在所有常见集群环境中运行，并能以内存速度和任意规模进行计算。 ??Apache Flink is a framework and distributed processing engine for stateful computations over unbounded and bounded data streams. Flink has been designed to run in all common cluster environments, perform computations at in-memory speed and at any scale.

??3) 有界流和无界流

??可以看出，状态流计算将逐渐成为企业构建数据平台的架构模式。目前，从社区的角度来看，只能满足Apache Flink。Flink通过实现Google Dataflow实现高吞吐、低延迟、高性能的流式计算框架。同时Flink支持高度容错的状态管理，防止因系统异常而丢失状态，Flink通过分布式快照技术周期性地进行Checkpoints即使在系统停机或异常情况下，实现状态的持久维护也能计算出正确的结果。 ??自 2019 年 1 从月起，阿里巴巴逐渐维护内部 Blink 回馈给 Flink 目前，开源社区贡献的代码数量已超过 100 万行。国内包括腾讯、百度、字节跳动等公司，国外包括 Uber、Lyft、Netflix 等公司都是 Flink 的使用者。

??在实际生产过程中，金融交易数据、互联网订单数据、GPS定位数据、传感器信号、移动终端生成的数据、通信信号数据、熟悉的网络流量监控和服务器生成的日志数据。这些数据最大的共同点是实时从不同的数据源中生成，然后传输到下游的分析系统。这些数据类型主要包括实时智能推荐、复杂事件处理、实时欺诈检测、实时数字仓库和ETL实时业务场景，如类型、实时报表类型等实时业务场景Flink对这类场景有很好的支持。

??（一）实时智能推荐

??根据用户历史的购买行为，智能推荐将通过推荐算法培训模型预测用户未来可能购买的物品。对于个人来说，推荐系统起着信息过滤的作用Web/App在服务方面，推荐系统起到满足用户个性化需求、提高用户满意度的作用。除了算法越来越完善外，推荐系统本身也在迅速发展，对延迟的要求也越来越苛刻和实时。利用Flink流量计算帮助用户构建更实时的智能推荐系统，实时计算用户行为指标，实时更新模型，实时预测用户指标，并推送预测信息Wep/App另一方面，帮助用户获取所需的商品信息，也帮助企业增加销量，创造更大的商业价值。

??(二)处理复杂事件

??对于复杂的事件处理，常见的案例主要集中在工业领域，如车载传感器、机械设备等实时故障检测。这些业务类型通常有大量的数据，对数据处理的及时性有很高的要求。通过利用Flink提供的CEP(复杂事件处理)提取事件模式，同时应用Flink的Sql一旦事件触发报警规则，立即将报警结果传输到下游通知系统，实现设备故障的快速预警和监控，辆状态监控等目的。

  （三）实时欺诈检测

  在金融领域的业务中，常常出现各种类型的欺诈行为，例如信用卡欺诈、信贷申请欺诈等，而如何保证用户和公司的资金安全，是来近年来许多金融公司及银行共同面对的挑战。随着不法分子欺诈手段的不断升级，传统的反欺诈手段已经不足以解决目前所面临的问题。以往可能需要几个小时才能通过交易数据计算出用户的行为指标，然后通过规则判别出具有欺诈行为嫌疑的用户，再进行案件调查处理，在这种情况下资金可能早已被不法分子转移，从而给企业和用户造成大量的经济损失。而运用Flink流式计算技术能够在毫秒内就完成对欺诈判断行为指标的计算，然后实时对交易流水进行规则判断或者模型预测，这样一旦检测出交易中存在欺诈嫌疑，则直接对交易进行实时拦截，避免因为处理不及时而导致的经济损失。

  （四）实时数仓与ETL

  结合离线数仓，通过利用流计算诸多优势和SQL灵活的加工能力，对流式数据进行实时清洗、归并、结构化处理，为离线数仓进行补充和优化。另一方面结合实时数据ETL处理能力，利用有状态流式计算技术，可以尽可能降低企业由于在离线数据计算过程中调度逻辑的复杂度，高效快速地处理企业需要的统计结果，帮助企业更好地应用实时数据所分析出来的结果。

  （五）流数据分析

  实时计算各类数据指标，并利用实时结果及时调整在线系统相关策略，在各类内容投放、无线智能推送领域有大量的应用。流式计算技术将数据分析场景实时化，帮助企业做到实时化分析Web应用或者App应用的各项指标，包括App版本分布情况、Crash检测和分布等，同时提供多维度用户行为分析，支持日志自主分析，助力开发者实现基于大数据技术的精细化运营、提升产品质量和体验、增强用户黏性。

  （六）实时报表分析

  实时报表分析是近年来很多公司采用的报表统计方案之一，其中最主要的应用便是实时大屏展示。利用流式计算实时得出的结果直接被推送到前端应用，实时显示出重要指标的变换情况。最典型的案例便是淘宝的双十一活动，每年双十一购物节，除疯狂购物外，最引人注目的就是天猫双十一大屏不停跳跃的成交总额。在整个计算链路中包括从天猫交易下单购买到数据采集、数据计算、数据校验，最终落到双十一大屏上展现的全链路时间压缩在5秒以内，顶峰计算性能高达数三十万笔订单/秒，通过多条链路流计算备份确保万无一失。而在其他行业，企业也在构建自己的实时报表系统，让企业能够依托于自身的业务数据，快速提取出更多的数据价值，从而更好地服务于企业运行过程中。

 1) Flink 的具体优势和特点有以下几点

 （一）同时支持高吞吐、低延迟、高性能

  Flink是目前开源社区中唯一一套集高吞吐、低延迟、高性能三者于一身的分布式流式数据处理框架。像Apache Spark也只能兼顾高吞吐和高性能特性，主要因为在Spark Streaming流式计算中无法做到低延迟保障；而流式计算框架Apache Storm只能支持低延迟和高性能特性，但是无法满足高吞吐的要求。而满足高吞吐、低延迟、高性能这三个目标对分布式流式计算框架来说是非常重要的。

 （二）支持事件时间（Event Time）概念

  在流式计算领域中，窗口计算的地位举足轻重，但目前大多数框架窗口计算采用的都是系统时间（Process Time），也是事件传输到计算框架处理时，系统主机的当前时间。Flink能够支持基于事件时间（Event Time）语义进行窗口计算，也就是使用事件产生的时间，这种基于事件驱动的机制使得事件即使乱序到达，流系统也能够计算出精确的结果，保持了事件原本产生时的时序性，尽可能避免网络传输或硬件系统的影响。

 （三）支持有状态计算

  Flink在1.4版本中实现了状态管理，所谓状态就是在流式计算过程中将算子的中间结果数据保存在内存或者文件系统中，等下一个事件进入算子后可以从之前的状态中获取中间结果中计算当前的结果，从而无须每次都基于全部的原始数据来统计结果，这种方式极大地提升了系统的性能，并降低了数据计算过程的资源消耗。对于数据量大且运算逻辑非常复杂的流式计算场景，有状态计算发挥了非常重要的作用。

 （四）支持高度灵活的窗口（Window）操作

  在流处理应用中，数据是连续不断的，需要通过窗口的方式对流数据进行一定范围的聚合计算，例如统计在过去的1分钟内有多少用户点击某一网页，在这种情况下，我们必须定义一个窗口，用来收集最近一分钟内的数据，并对这个窗口内的数据进行再计算。Flink将窗口划分为基于Time、Count、Session，以及Data-driven等类型的窗口操作，窗口可以用灵活的触发条件定制化来达到对复杂的流传输模式的支持，用户可以定义不同的窗口触发机制来满足不同的需求。

 （五）基于轻量级分布式快照（CheckPoint）实现的容错

  Flink能够分布式运行在上千个节点上，将一个大型计算任务的流程拆解成小的计算过程，然后将tesk分布到并行节点上进行处理。在任务执行过程中，能够自动发现事件处理过程中的错误而导致数据不一致的问题，比如：节点宕机、网路传输问题，或是由于用户因为升级或修复问题而导致计算服务重启等。在这些情况下，通过基于分布式快照技术的Checkpoints，将执行过程中的状态信息进行持久化存储，一旦任务出现异常停止，Flink就能够从Checkpoints中进行任务的自动恢复，以确保数据在处理过程中的一致性（Exactly-Once）。

 （六）基于JVM实现独立的内存管理

  内存管理是所有计算框架需要重点考虑的部分，尤其对于计算量比较大的计算场景，数据在内存中该如何进行管理显得至关重要。针对内存管理，Flink实现了自身管理内存的机制，尽可能减少JVM GC对系统的影响。另外，Flink通过序列化/反序列化方法将所有的数据对象转换成二进制在内存中存储，降低数据存储的大小的同时，能够更加有效地对内存空间进行利用，降低GC带来的性能下降或任务异常的风险，因此Flink较其他分布式处理的框架会显得更加稳定，不会因为JVM GC等问题而影响整个应用的运行。

 （七）Save Points（保存点）

  对于7*24小时运行的流式应用，数据源源不断地接入，在一段时间内应用的终止有可能导致数据的丢失或者计算结果的不准确，例如进行集群版本的升级、停机运维操作等操作。值得一提的是，Flink通过Save Points技术将任务执行的快照保存在存储介质上，当任务重启的时候可以直接从事先保存的Save Points恢复原有的计算状态，使得任务继续按照停机之前的状态运行，Save Points技术可以让用户更好地管理和运维实时流式应用。

 2) 流式计算框架的对比

  Storm是比较早的流式计算框架，后来又出现了Spark Streaming和Trident，现在又出现了Flink这种优秀的实时计算框架，那么这几种计算框架到底有什么区别呢？

• 模型：Storm和Flink是真正的一条一条处理数据；而Trident（Storm的封装框架）和Spark Streaming其实都是小批处理，一次处理一批数据（小批量）。
• API：Storm和Trident都使用基础API进行开发，比如实现一个简单的sum求和操作；而Spark Streaming和Flink中都提供封装后的高阶函数，可以直接来使用，非常方便。
• 保证次数：在数据处理方面，Storm可以实现至少处理一次，但不能保证仅处理一次，这样就会导致数据重复处理问题，所以针对计数类的需求，可能会产生一些误差；Trident通过事务可以保证对数据实现仅一次的处理，Spark Streaming和Flink也是如此。
• 容错机制：Storm和Trident可以通过ACK机制实现数据的容错机制，而Spark Streaming和Flink可以通过CheckPoint机制实现容错机制。
• 状态管理：Storm中没有实现状态管理，Spark Streaming实现了基于DStream的状态管理，而Trident和Flink实现了基于操作的状态管理。
• 延时：表示数据处理的延时情况，因此Storm和Flink接收到一条数据就处理一条数据，其数据处理的延时性是很低的；而Trident和Spark Streaming都是小型批处理，它们数据处理的延时性相对会偏高。
• 吞吐量：Storm的吞吐量其实也不低，只是相对于其他几个框架而言较低；Trident属于中等；而Spark Streaming和Flink的吞吐量是比较高的。

 Flink课程选择的是Apache Flink 1.9.1 版本，是目前最新的稳定版本，并且兼容性比较好。下载地址：  https://flink.apache.org/zh/downloads.html

 1) 开发工具

  先说明一下开发工具的问题。官方建议使用IntelliJ IDEA，因为它默认集成了Scala和Maven环境，使用更加方便，当然使用Eclipse也是可以的。我们这门课使用IDEA。开发Flink程序时，可以使用Java、Python或者Scala语言，本课程全部使用Scala，因为使用Scala实现函数式编程会比较简洁。学生可以在课后自己补充JAVA代码。

 2) 配置依赖

  创建项目，并且修改源代码目录为scala 案例需求：采用Netcat 数据源发送数据，使用Flink统计每个单词的数量。 注意： Flink流式处理数据时，需要导入隐式转换：org.apache.flink.streaming.api.scala._

package com.bjsxt.flink

import org.apache.flink.streaming.api.scala.{ 
        DataStream, StreamExecutionEnvironment}

//基于流计算的WordCount案例
object StreamWordCount { 
        

  def main(args: Array[String]): Unit = { 
        

    //初始化Flink的Streaming（流计算）上下文执行环境
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //导入隐式转换，建议写在这里，可以防止IDEA代码提示出错的问题
    import org.apache.flink.streaming.api.scala._

    //读取数据
    val stream: DataStream[String] = streamEnv.socketTextStream("mynode5",8888)

    //转换计算
    val result: DataStream[(String, Int)] = stream.flatMap(_.split(","))
      .map((_, 1))
      .keyBy(0)
      .sum(1)

    //打印结果到控制台
    result.print()

    //启动流式处理，如果没有该行代码上面的程序不会运行
    streamEnv.execute("wordcount")
  }
}

 在Linux系统中使用nc命令发送数据测试

 nc  -lk  8888

需求：读取本地数据文件，统计文件中每个单词出现的次数。 根据需求，很明显是有界流（批计算），所以采用另外一个上下文环境：ExecutionEnvironment

package com.bjsxt.flink

import java.net.URL

import org.apache.flink.api.scala.ExecutionEnvironment

object BatchWordCount { 
        

  def main(args: Array[String]): Unit = { 
        
    //初始化flink的环境
    val env: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment

    //导入隐式转换，建议写在这里，可以防止IDEA代码提示出错的问题
    import org.apache.flink.api.scala._

    //读取数据
    val dataURL = getClass.getResource("/wc.txt")//wc.txt文件在main目录下的resources中
    val data: DataSet[String] = env.readTextFile(dataURL.getPath)

    //计算
    val result: AggregateDataSet[(String, Int)] = data.flatMap(_.split(" "))
      .map((_, 1))
      .groupBy(0)  //其中0代表元组中的下标，“0”下标代表：单词
      .sum(1)      //其中1代表元组中的下标，“1”下标代表：单词出现的次数

    //打印结果
    result.print()
  }
}

  Flink的安装和部署主要分为本地（单机）模式和集群模式，其中本地模式只需直接解压就可以使用，不以修改任何参数，一般在做一些简单测试的时候使用。本地模式在我们的课程里面不再赘述。集群模式包含：

• Standalone。
• Flink on Yarn。
• Mesos。
• Docker。
• Kubernetes。
• AWS。
• Goole Compute Engine。

目前在企业中使用最多的是Flink on Yarn模式。我们的课程中讲Standalone和Flink on Yarn这两种模式。

 Flink整个系统主要由两个组件组成，分别为JobManager和TaskManager，Flink架构也遵循Master-Slave架构设计原则，JobManager为Master节点，TaskManager为Worker（Slave）节点。所有组件之间的通信都是借助于Akka Framework，包括任务的状态以及Checkpoint触发等信息。  1) Client客户端

  客户端负责将任务提交到集群，与JobManager构建Akka连接，然后将任务提交到JobManager，通过和JobManager之间进行交互获取任务执行状态。客户端提交任务可以采用CLI方式或者通过使用Flink WebUI提交，也可以在应用程序中指定JobManager的RPC网络端口构建ExecutionEnvironment提交Flink应用。

 2) JobManager

  JobManager负责整个Flink集群任务的调度以及资源的管理，从客户端中获取提交的应用，然后根据集群中TaskManager上TaskSlot的使用情况，为提交的应用分配相应的TaskSlots资源并命令TaskManger启动从客户端中获取的应用。JobManager相当于整个集群的Master节点，且整个集群中有且仅有一个活跃的JobManager，负责整个集群的任务管理和资源管理。JobManager和TaskManager之间通过Actor System进行通信，获取任务执行的情况并通过Actor System将应用的任务执行情况发送给客户端。同时在任务执行过程中，Flink JobManager会触发Checkpoints操作，每个TaskManager节点收到Checkpoint触发指令后，完成Checkpoint操作，所有的Checkpoint协调过程都是在Flink JobManager中完成。当任务完成后，Flink会将任务执行的信息反馈给客户端，并且释放掉TaskManager中的资源以供下一次提交任务使用。

 3) TaskManager

  TaskManager相当于整个集群的Slave节点，负责具体的任务执行和对应任务在每个节点上的资源申请与管理。客户端通过将编写好的Flink应用编译打包，提交到JobManager，然后JobManager会根据已经注册在JobManager中TaskManager的资源情况，将任务分配给有资源的TaskManager节点，然后启动并运行任务。TaskManager从JobManager接收需要部署的任务，然后使用Slot资源启动Task，建立数据接入的网络连接，接收数据并开始数据处理。同时TaskManager之间的数据交互都是通过数据流的方式进行的。   可以看出，Flink的任务运行其实是采用多线程的方式，这和MapReduce多JVM进程的方式有很大的区别Fink能够极大提高CPU使用效率，在多个任务和Task之间通过TaskSlot方式共享系统资源，每个TaskManager中通过管理多个TaskSlot资源池进行对资源进行有效管理。

  Standalone是Flink的独立部署模式，它不依赖其他平台。在使用这种模式搭建Flink集群之前，需要先规划集群机器信息。在这里为了搭建一个标准的Flink集群，这里准备3台Linux机器，如图下所示。  1) 解压 Flink 的压缩包  2) 修改配置文件

  ①　进入到conf目录下，编辑flink-conf.yaml配置文件：   其中：taskmanager.numberOfTaskSlot 参数默认值为1，修改成3。表示数每一个TaskManager上有3个Slot。

  ②　编辑conf/slaves配置文件  3) 分发给另外两台服务器  4) 启动 Flink 集群服务

 5) 访问 WebUI  6) 通过命令提交 job 到集群

  ①　把上一章节中第一个 Flink 流处理案例代码打包，并上传   ②　执行命令： 在执行命令之前先确保 nc -lk 8888 是否启动 其中-d选项表示提交job之后，客户端结束并退出。之后输入测试数据 ③　查看job执行结果 然后去 hadoop101 的 TaskManager 上查看最后的结果：  7) 通过 WebUI 提交 job 到集群 注意：通过webui上传的jar包会默认放在web.tmpdir目录下，这个目录在/tmp/flink-web-UUID组成，可以在jobManager的webui中查看，每次集群重启后这个目录会被删除重建，可以修改这个目录保存之前上传的jar包。

 8) 配置文件参数说明

 下面针对 flink-conf.yaml 文件中的几个重要参数进行分析：

• jobmanager.heap.size：JobManager节点可用的内存大小。
• taskmanager.heap.size：TaskManager节点可用的内存大小。
• taskmanager.numberOfTaskSlots：每台机器可用的Slot数量。
• parallelism.default：默认情况下Flink任务的并行度。

 上面参数中所说的 Slot 和 parallelism 的区别：

• Slot是静态的概念，是指TaskManager具有的并发执行能力。
• parallelism是动态的概念，是指程序运行时实际使用的并发能力。
• 设置合适的parallelism能提高运算效率。

3. Flink提交到Yarn

 Flink on Yarn模式的原理是依靠YARN来调度Flink任务，目前在企业中使用较多。这种模式的好处是可以充分利用集群资源，提高集群机器的利用率，并且只需要1套Hadoop集群，就可以执行MapReduce和Spark任务，还可以执行Flink任务等，操作非常方便，不需要维护多套集群，运维方面也很轻松。Flink on Yarn模式需要依赖Hadoop集群，并且Hadoop的版本需要是2.2及以上。我们的课程里面选择的Hadoop版本是2.7.5。

 Flink On Yarn的内部实现原理：

• 当启动一个新的Flink YARN Client会话时，客户端首先会检查所请求的资源（容器和内存）是否可用。之后，它会上传Flink配置和JAR文件到HDFS。
• 客户端的下一步是请求一个YARN容器启动ApplicationMaster。JobManager和ApplicationMaster(AM)运行在同一个容器中，一旦它们成功地启动了，AM就能够知道JobManager的地址，它会为TaskManager生成一个新的Flink配置文件（这样它才能连上JobManager），该文件也同样会被上传到HDFS。另外，AM容器还提供了Flink的Web界面服务。Flink用来提供服务的端口是由用户和应用程序ID作为偏移配置的，这使得用户能够并行执行多个YARN会话。
• 之后，AM开始为Flink的TaskManager分配容器（Container），从HDFS下载JAR文件和修改过的配置文件。一旦这些步骤完成了，Flink就安装完成并准备接受任务了。

 Flink on Yarn模式在使用的时候又可以分为两种：

• 第1种模式(Session-Cluster)：是在YARN中提前初始化一个Flink集群(称为Flink yarn-session)，开辟指定的资源，以后的Flink任务都提交到这里。这个Flink集群会常驻在YARN集群中，除非手工停止（yarn application -kill id），当手动停止yarn application对应的id时，运行在当前application上的所有flink任务都会被kill。这种方式创建的Flink集群会独占资源，不管有没有Flink任务在执行，YARN上面的其他任务都无法使用这些资源。

• 第2种模式(Per-Job-Cluster)：每次提交Flink任务都会创建一个新的Flink集群，每个Flink任务之间相互独立、互不影响，管理方便。任务执行完成之后创建的Flink集群也会消失，不会额外占用资源，按需使用，这使资源利用率达到最大，在工作中推荐使用这种模式。当杀掉一个当前yarn flink任务时，不会影响其他flink任务执行。

注意：Flink on Yarn 还需要以下先决条件：

• 配置Hadoop的环境变量
• 关闭yarn的虚拟内存检查。

在每台 nodemanager 节点上 $HADOOP_HOME/etc/hadoop/yarn-site.xml中配置如下配置：

<property>
 <name>yarn.nodemanager.vmem-check-enabled</name>
   <value>false</value>
 </property>

• 下载Flink提交到Hadoop的连接器（jar包），并把jar拷贝到Flink的lib目录下 注意：可以将这个jar包拷贝到所有的flink节点上lib目录下，也可以只是拷贝到对应的提交任务的flink节点上。

 1) Session-Cluster 模式（yarn-session）

 ①　先启动Hadoop集群，然后通过命令启动一个Flink的yarn-session集群：

  bin/yarn-session.sh  -n 3 -s 3 -nm bjsxt  -d

其中yarn-session.sh后面支持多个参数。下面针对一些常见的参数进行讲解：

• -n,–container <arg> 表示分配容器的数量（也就是TaskManager的数量）。目前版本yarn-session.sh提交任务中，经过测试发现，-n参数是不起作用的。
• -D <arg> 动态属性。
• -d,–detached在后台独立运行。
• -jm,–jobManagerMemory <arg>：设置JobManager的内存，单位是MB。
• -nm,–name：在YARN上为一个自定义的应用设置一个名字。
• -q,–query：显示YARN中可用的资源（内存、cpu核数）。
• -qu,–queue <arg>：指定YARN队列。
• -s,–slots <arg>：每个TaskManager使用的Slot数量。
• -tm,–taskManagerMemory <arg>：每个TaskManager的内存，单位是MB。
• -z,–zookeeperNamespace <arg>：针对HA模式在ZooKeeper上创建NameSpace。
• -id,–applicationId <yarnAppId>：指定YARN集群上的任务ID，附着到一个后台独立运行的yarn session中。

 ②　查看WebUI: 由于还没有提交Flink job，所以都是0。 这个时候注意查看本地文件系统中有一个临时文件。有了这个文件可以提交 job 到 Yarn  ③　提交Job : 由于有了之前的配置，所以自动会提交到Yarn中。

  bin/flink run -c com.bjsxt.flink.StreamWordCount /home/Flink-Demo-1.0-SNAPSHOT.jar

 注意：如果删除目录/tmp/.yarn-properties-root文件，那么再按照以上命令提交任务，将会是寻找Standalone模式中的jobManager节点提交，如果想要重新提交到当前yarn-session中可以使用-yid命令指定对应的yarn application的id，命令如下：

./flink run -yid  application_1598346048136_0002  -c com.lw.scala.myflink.streaming.example.FlinkReadSocketData  /root/test/MyFlink-1.0-SNAPSHOT-jar-with-dependencies.jar

 至此第一种模式全部完成。

2) Pre-Job-Cluster 模式（yarn-cluster）

 这种模式下不需要先启动yarn-session。所以我们可以把前面启动的yarn-session集群先停止，停止的命令是:

yarn application -kill application_1576832892572_0002
//其中 application_1576832892572_0002 是ID

 确保Hadoop集群是健康的情况下直接提交Job命令：

bin/flink run -m yarn-cluster -yn 3 -ys 3 -ynm bjsxt02 -c com.bjsxt.flink.StreamWordCount /home/Flink-Demo-1.0-SNAPSHOT.jar

 可以看到一个全新的yarn-session 任务提交参数讲解：相对于Yarn-Session参数而言，只是前面加了y。

• -yn,–container <arg> 表示分配容器的数量，也就是TaskManager的数量。目前版本yarn-cluster提交任务中，经过测试发现，-yn参数是不起作用的。
• -d,–detached：设置在后台运行。
• -yjm,–jobManagerMemory<arg>:设置JobManager的内存，单位是MB。
• -ytm，–taskManagerMemory<arg>:设置每个TaskManager的内存，单位是MB。
• -ynm,–name:给当前Flink application在Yarn上指定名称。
• -yq,–query：显示yarn中可用的资源（内存、cpu核数）
• -yqu,–queue<arg> :指定yarn资源队列
• -ys,–slots<arg> :每个TaskManager使用的Slot数量。
• -yz,–zookeeperNamespace<arg>:针对HA模式在Zookeeper上创建NameSpace
• -yid,–applicationID<yarnAppId> : 指定Yarn集群上的任务ID,附着到一个后台独立运行的Yarn Session中。

 默认情况下，每个Flink集群只有一个JobManager，这将导致单点故障（SPOF），如果这个JobManager挂了，则不能提交新的任务，并且运行中的程序也会失败。使用JobManager HA，集群可以从JobManager故障中恢复，从而避免单点故障。用户可以在Standalone或Flink on Yarn集群模式下配置Flink集群HA（高可用性）。  Standalone模式下，JobManager的高可用性的基本思想是，任何时候都有一个Alive JobManager和多个Standby JobManager。Standby JobManager可以在Alive JobManager挂掉的情况下接管集群成为Alive JobManager，这样避免了单点故障，一旦某一个Standby JobManager接管集群，程序就可以继续运行。Standby JobManagers和Alive JobManager实例之间没有明确区别，每个JobManager都可以成为Alive或Standby。  1) Flink Standalone集群的HA安装和配置(目前测试1.9版本有bug)

  实现HA还需要依赖ZooKeeper和HDFS，因此要有一个ZooKeeper集群和Hadoop集群，首先启动Zookeeper集群和HDFS集群。我们的课程中分配3台JobManager，如下表：

  ①　修改配置文件conf/masters   ②　修改配置文件conf/flink-conf.yaml

#要启用高可用，设置修改为zookeeper
high-availability: zookeeper
#Zookeeper的主机名和端口信息，多个参数之间用逗号隔开
high-availability.zookeeper.quorum: hadoop103:2181,hadoop101:2181,hadoop102:2181
# 建议指定HDFS的全路径。如果某个Flink节点没有配置HDFS的话，不指定HDFS的全路径则无法识到，storageDir存储了恢复一个JobManager所需的所有元数据。这里如果指定hdfs路径需要在每台节点上配置hadoop的依赖包flink-shaded-hadoop-2-uber-2.7.5-10.0.jar。
high-availability.storageDir: hdfs://mycluster/flink/ha

  ③　把修改的配置文件拷贝其他服务器中

[root@hadoop101 ]# scp -r ./flink-xxx root@hadoop102:`pwd`
[root@hadoop101 ]# scp -r ./flink-xxx root@hadoop103:`pwd`

  ④　启动集群 版本问题：目前使用Flink1.7.1版本测试没有问题，使用Flink1.9版本存在HA界面不能自动跳转到对应的Alive JobManager的现象。

 2) Flink On Yarn HA 安装和配置

  正常基于Yarn提交Flink程序，无论是使用yarn-session模式还是yarn-cluster模式，基于yarn运行后的application 只要kill 掉对应的Flink 集群进程“YarnSessionClusterEntrypoint”后，基于Yarn的Flink任务就失败了，不会自动进行重试，所以基于Yarn运行Flink任务，也有必要搭建HA，这里同样还是需要借助zookeeper来完成，步骤如下：

  ①　修改所有Hadoop节点的yarn-site.xml

  将所有Hadoop节点的yarn-site.xml中的提交应用程序最大尝试次数调大,这里默认是2次，也可以不调。

#在每台hadoop节点yarn-site.xml中设置提交应用程序的最大尝试次数，建议不低于4，这里重试指的是ApplicationMaster
<property>
  <name>yarn.resourcemanager.am.max-attempts</name>
  <value>4</value>
</property>

  ②　启动Hadoop集群   启动zookeeper，启动Hadoop集群。   ③　修改Flink对应flink-conf.yaml配置   配置对应的conf下的flink-conf.yaml，配置内容如下：

#配置依赖zookeeper模式进行HA搭建
high-availability: zookeeper
#配置JobManager原数据存储路径
high-availability.storageDir: hdfs://mycluster/flink/yarnha/
#配置zookeeper集群节点
high-availability.zookeeper.quorum: hadoop101:2181,hadoop102:2181,hadoop103:2181
#向yarn提交一个application重试的次数，也可以不设置。
yarn.application-attempts: 10

  ④　启动yarn-session.sh 测试HA： yarn-session.sh -n 2 ，也可以直接提交Job 启动之后，可以登录yarn中对应的flink webui，如下图示：   点击对应的Tracking UI，进入Flink 集群UI:   查看对应的JobManager在哪台节点上启动： 进入对应的节点，kill掉对应的“YarnSessionClusterEntrypoint”进程。然后进入到Yarn中观察“applicationxxxx_0001”job信息： 点击job ID,发现会有对应的重试