大数据Hadoop之——实时计算流计算引擎Flink（Flink环境部署）

大数据Hadoop之——实时计算流计算引擎Flink（Flink环境部署）

@[TOC]

一、概述

三、Flink核心概念

1）Time（时间语义）

Flink 中的 Time 分为三种：事件时间、达到时间与处理时间。 事件时间：是事件真实发生的时间。

达到时间：是系统接收到事件的时间，即服务端接收到事件的时间。 处理时间：是系统开始处理到达事件的时间。

【温馨提示】在某些场景下，处理时间等于达到时间。因为处理时间没有乱序的问题，所以服务端做基于处理时间的计算是比较简单的，无迟到与乱序数据。

Flink 中只需要通过 env 环境变量即可设置Time：

//创建环境上下文 val env = StreamExecutionEnvironment.getExecutionEnvironment // 设置在当前程序中使用 ProcessingTime env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

2）Window（窗口）

窗口本质就是将无限数据集沿着时间（或者数量）的边界切分成有限数据集。 Time Window：基于时间的，分为Tumbling Window（无数据重叠）和Sliding Window（有数据重叠） 。

Count Window：基于数量的，分为Tumbling Window（无数据重叠）和Sliding Window（有数据重叠）。 Session Window：基于会话的，一个session window关闭通常是由于一段时间没有收到元素。 Global Window：全局窗口。

【温馨提示】在实际操作中，window又分为两大类型的窗口：Keyed Window 和 Non-keyed Window，两种类型的窗口操作API有细微的差别。

3） Trigger

1、自定义触发器

触发器决定了窗口何时会被触发计算，Flink 中开发人员需要在 window 类型的操作之后才能调用 trigger 方法传入触发器定义。Flink 中的触发器定义需要继承并实现 Trigger 接口，该接口有以下方法：

onElement()： 每个被添加到窗口中的元素都会被调用 onEventTime()：当事件时间定时器触发时会被调用，比如watermark到达 onProcessingTime()：当处理时间定时器触发时会被调用，比如时间周期触发 onMerge()：当两个窗口合并时两个窗口的触发器状态将会被调动并合并 clear()：执行需要清除相关窗口的事件

以上方法会返回决定如何触发执行的 TriggerResult：

CONTINUE：什么都不做 FIRE：触发计算 PURGE：清除窗口中的数据 FIRE_AND_PURGE：触发计算后清除窗口中的数据 2、预定义触发器 如果开发人员未指定触发器，则 Flink 会自动根据场景使用默认的预定义好的触发器。在基于事件时间的窗口中使用 EventTimeTrigger，该触发器会在watermark通过窗口边界后立即触发（即watermark出现关闭改窗口时）。在全局窗口（GlobalWindow）中使用 NeverTrigger，该触发器永远不会触发，所以在使用全局窗口时用户需要自定义触发器。

4）State

Managed State 是由flink runtime管理来管理的，自动存储、自动恢复，在内存管理上有优化机制。且Managed State 支持常见的多种数据结构，如value、list、map等，在大多数业务场景中都有适用之处。总体来说是对开发人员来说是比较友好的，因此 Managed State 是 Flink 中最常用的状态。Managed State 又分为 Keyed State 和 Operator State 两种。 Raw State 由用户自己管理，需要序列化，只能使用字节数组的数据结构。Raw State 的使用和维度都比 Managed State 要复杂，建议在自定义的Operator场景中酌情使用。

5）状态存储

Flink中状态的实现有三种：MemoryState、FsState、RocksDBState。三种状态存储方式与使用场景各不相同，详细介绍如下：

1、MemoryStateBackend

构造函数：MemoryStateBackend(int maxStateSize, boolean asyncSnapshot) 存储方式：State存储于各个 TaskManager内存中，Checkpoint存储于 JobManager内存 容量限制：单个State最大5M、maxStateSize<=akka.framesize（10M）、总大小不超过JobManager内存 使用场景：无状态或者JobManager挂掉不影响的测试环境等，不建议在生产环境使用

2、FsStateBackend

构造函数：FsStateBackend(URI checkpointUri, boolean asyncSnapshot) 存储方式：State存储于 TaskManager内存，Checkpoint存储于 外部文件系统（本次磁盘 or HDFS） 容量限制：State总量不超过TaskManager内存、Checkpoint总大小不超过外部存储空间 使用场景：常规使用状态的作业，分钟级的窗口聚合等，可在生产环境使用

3、RocksDBStateBackend

构造函数：RocksDBStateBackend(URI checkpointUri, boolean enableincrementCheckpoint) 存储方式：State存储于 TaskManager上的kv数据库（内存+磁盘），Checkpoint存储于 外部文件系统（本次磁盘 or HDFS） 容量限制：State总量不超过TaskManager内存+磁盘、单key最大2g、Checkpoint总大小不超过外部存储空间 使用场景：超大状态的作业，天级的窗口聚合等，对读写性能要求不高的场景，可在生产环境使用

根据业务场景需要用户选择最合适的 StateBackend ，代码中只需在相应的 env 环境中设置即可：

// flink 上下文环境变量 val env = StreamExecutionEnvironment.getExecutionEnvironment // 设置状态后端为 FsStateBackend，数据存储到 hdfs /tmp/flink/checkpoint/test 中 env.setStateBackend(new FsStateBackend("hdfs://ns1/tmp/flink/checkpoint/test", false))

6）Checkpoint

Checkpoint 是分布式全域一致的，数据会被写入hdfs等共享存储中。且其产生是异步的，在不中断、不影响运算的前提下产生。

用户只需在相应的 env 环境中设置即可：

// 1000毫秒进行一次 Checkpoint 操作 env.enableCheckpointing(1000) // 模式为准确一次 env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE) // 两次 Checkpoint 之间最少间隔 500毫秒 env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500) // Checkpoint 过程超时时间为 60000毫秒，即1分钟视为超时失败 env.getCheckpointConfig.setCheckpointTimeout(60000) // 同一时间只允许1个Checkpoint的操作在执行 env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)

1、Asynchronous Barrier Snapshots(ABS)

异步屏障快照算法，这个算法基本上是Chandy-Lamport算法的变体，针对DAG（有向无环图）的ABS算法执行流程如下所示：

Barrier周期性的被注入到所有的Source中，Source节点看到Barrier后，会立即记录自己的状态，然后将Barrier发送到Transformation Operator。 当Transformation Operator从某个input channel收到Barrier后，它会立刻Block住这条通道，直到所有的input channel都收到Barrier，这个等待的过程就叫做屏障对齐（barrier alignment），此时该Operator就会记录自身状态，并向自己的所有output channel广播Barrier。 Sink接受Barrier的操作流程与Transformation Oper一样。当所有的Barrier都到达Sink之后，并且所有的Sink也完成了Checkpoint，这一轮Snapshot就完成了。

2、Exactly-Once vs At-Least-Once

上面讲到的屏障对齐过程是Flink exactly-once语义的基础，因为屏障对齐能够保证多输入流的算子正常处理不同checkpoint区间的数据，避免它们发生交叉，即不会有数据被处理两次。 但是对齐过程需要时间，有一些对延迟特别敏感的应用可能对准确性的要求没有那么高。所以Flink也允许在StreamExecutionEnvironment.enableCheckpointing()方法里指定At-Least-Once语义，会取消屏障对齐，即算子收到第一个输入的屏障之后不会阻塞，而是触发快照。这样一来，部分属于检查点n + 1的数据也会包括进检查点n的数据里， 当恢复时，这部分交叉的数据就会被重复处理。

7）Watermark

Flink 程序并 不能自动提取数据源中哪个字段/标识为数据的事件时间，从而也就无法自己定义 Watermark 。开发人员需要通过 Flink 提供的 API 来 提取和定义 Timestamp/Watermark，可以在 数据源或者数据流中 定义。

1、自定义数据源设置 Timestamp/Watermark

自定义的数据源类需要继承并实现 SourceFunction[T] 接口，其中 run 方法是定义数据生产的地方：

//自定义的数据源为自定义类型MyType class MySource extends SourceFunction[MyType]{ //重写run方法，定义数据生产的逻辑 override def run(ctx: SourceContext[MyType]): Unit = { while (/* condition */) { val next: MyType = getNext() //设置timestamp从MyType的哪个字段获取(eventTimestamp) ctx.collectWithTimestamp(next, next.eventTimestamp) if (next.hasWatermarkTime) { //设置watermark从MyType的那个方法获取(getWatermarkTime) ctx.emitWatermark(new Watermark(next.getWatermarkTime)) } } } }

2、在数据流中设置 Timestamp/Watermark

在这里插入代码片在数据流中，可以设置 stream 的 Timestamp Assigner ，该 Assigner 将会接收一个 stream，并生产一个带 Timestamp和Watermark 的新 stream。

val withTimestampsAndWatermarks: DataStream[MyEvent] = stream.assignTimestampsAndWatermarks(new MyTimestampsAndWatermarks())

8）广播状态（Broadcast State）

和 Spark 中的广播变量一样，Flink 也支持在各个节点中各存一份小数据集，所在的计算节点实例可在本地内存中直接读取被广播的数据，可以避免Shuffle提高并行效率。 广播状态（Broadcast State）的引入是为了支持一些来自一个流的数据需要广播到所有下游任务的情况，它存储在本地，用于处理其他流上的所有传入元素。

// key the shapes by color KeyedStream colorPartitionedStream = shapeStream.keyBy(new KeySelector(){...}); // a map descriptor to store the name of the rule (string) and the rule itself. MapStateDescriptor ruleStateDescriptor = new MapStateDescriptor<>("RulesBroadcastState",BasicTypeInfo.STRING_TYPE_INFO, TypeInformation.of(new TypeHint() {})); // broadcast the rules and create the broadcast state BroadcastStream ruleBroadcastStream = ruleStream.broadcast(ruleStateDescriptor); DataStream output = colorPartitionedStream.connect(ruleBroadcastStream).process(new KeyedBroadcastProcessFunction(){...});

9）Operator Chain

Flink作业中，可以指定相关的chain将相关性非常强的转换操作（operator）绑定在一起，使得上下游的Task在同一个Pipeline中执行，避免因为数据在网络或者线程之间传输导致的开销。 一般情况下Flink在Map类型的操作中默认开启 Operator Chain 以提高整体性能，开发人员也可以根据需要创建或者禁止 Operator Chain 对任务进行细粒度的链条控制。

//创建 chain dataStream.filter(...).map(...).startNewChain().map(...) //禁止 chain dataStream.map(...).disableChaining()

创建的链条只对当前的操作符和之后的操作符有效，不不影响其他操作，如上代码只针对两个map操作进行链条绑定，对前面的filter操作无效，如果需要可以在filter和map之间使用 startNewChain方法即可。

10）Side Output

除了从DataStream操作的结果中获取主数据流之外，Flink还可以产生任意数量额外的侧输出(Side Output)结果流。侧输出结果流的数据类型不需要与主数据流的类型一致，不同侧输出流的类型也可以不同。当要拆分数据流时(通常必须复制流)，从每个流过滤出不想拥有的数据时Side Output将非常有用。

DataStream input = ...; final OutputTag outputTag = new OutputTag("side-output"){}; SingleOutputStreamOperator mainDataStream = input .process(new ProcessFunction() { @Override public void processElement( Integer value, Context ctx, Collector out) throws Exception { // 将数据发送到常规输出中 out.collect(value); // 将数据发送到侧输出中 ctx.output(outputTag, "sideout-" + String.valueOf(value)); } }); DataStream sideOutputStream = mainDataStream.getSideOutput(outputTag);

四、对比常用的实时计算框架

Flink 是有状态的和容错的，可以在维护一次应用程序状态的同时无缝地从故障中恢复。 它支持大规模计算能力，能够在数千个节点上并发运行。 它具有很好的吞吐量和延迟特性。 同时，Flink 提供了多种灵活的窗口函数。 Flink 在流式计算里属于真正意义上的单条处理，每一条数据都触发计算，而不是像 Spark 一样的 Mini Batch 作为流式处理的妥协。 Flink的容错机制较为轻量，对吞吐量影响较小，而且拥有图和调度上的一些优化，使得 Flink 可以达到很高的吞吐量。 而 Strom 的容错机制需要对每条数据进行ack，因此其吞吐量瓶颈也是备受诟病。

五、Flink环境部署（Flink On Yarn）

下载地址：Local 模式来说，JobManager 和 TaskManager 会公用一个 JVM 来完成 Workload。如果要验证一个简单的应用，Local 模式是最方便的。实际企业中大多使用Flink On Yarn模式，而local模式只是将安装包解压启动（./bin/start-cluster.sh）即可。其实local模式就是单节点，master和woker节点都是同一台机器。

Local Cluster模式是开箱即用的，直接解压安装包，然后启动即可。

$ cd /opt/bigdata/hadoop/software $ wget https://dlcdn.apache.org/flink/flink-1.14.2/flink-1.14.2-bin-scala_2.12.tgz # 解压 $ tar -zxvf flink-1.14.2-bin-scala_2.12.tgz -C /opt/bigdata/hadoop/server/ # 进入bin目录运行启动脚本 $ cd /opt/bigdata/hadoop/server/flink-1.14.2 $ ./bin/start-cluster.sh

2）Standalone模式

Stanalone CLuster是一种独立的集群模式，集群运行不需要依赖外部系统，完全自己独立进行管理。

1、机器及角色划分

1、下载

$ cd /opt/bigdata/hadoop/software $ wget https://dlcdn.apache.org/flink/flink-1.14.2/flink-1.14.2-bin-scala_2.12.tgz # 解压 $ tar -zxvf flink-1.14.2-bin-scala_2.12.tgz -C /opt/bigdata/hadoop/server/ $ cd /opt/bigdata/hadoop/server/flink-1.14.2

2、修改配置文件

修改flink-conf.yaml文件 $ cd /opt/bigdata/hadoop/server/flink-1.14.2/conf $ vi flink-conf.yaml

jobmanager节点地址，也是master节点地址

jobmanager.rpc.address: hadoop-node1

> **其它使用默认配置**，其中有一些HA高可用、容错、安全、HistoryServer相关配置，按需进行配置即可，HistoryServer需单独运行启动脚本来启动服务。 - **修改masters文件** 把默认的localhost:8081删掉，添加如下内容： ```bash hadoop-node1:8081

修改workers文件，内容如下：把默认的localhost删掉，添加如下内容： hadoop-node2 hadoop-node3

3、将安装目录复制到另外两台节点

$ cd /opt/bigdata/hadoop/server $ scp -r flink-1.14.2 hadoop-node2:/opt/bigdata/hadoop/server/ $ scp -r flink-1.14.2 hadoop-node3:/opt/bigdata/hadoop/server/

4、配置环境变量，修改/etc/profile

在/etc/profile文件中添加如下内容（所有节点）：

export FLINK_HOME=/opt/bigdata/hadoop/server/flink-1.14.2 export PATH=$PATH:$FLINK_HOME/bin

使配置文件生效

$ source /etc/profile

5、启动集群

$ start-cluster.sh

3）On Yarn模式（推荐）

YARN模式是使用YARN做为Flink运行平台，JobManager、TaskManager、用户提交的应用程序都运行在YARN上。

FLink on yarn 有三种运行模式：

yarn-session模式（Seesion Mode） yarn-cluster模式（Per-Job Mode） Application模式（Application Mode）

下载

$ cd /opt/bigdata/hadoop/software $ wget https://dlcdn.apache.org/flink/flink-1.14.2/flink-1.14.2-bin-scala_2.12.tgz # 解压 $ tar -zxvf flink-1.14.2-bin-scala_2.12.tgz -C /opt/bigdata/hadoop/server/ $ cd /opt/bigdata/hadoop/server/flink-1.14.2

export FLINK_HOME=/opt/bigdata/hadoop/server/flink-1.14.2 export PATH=$PATH:$FLINK_HOME/bin # 上面两句如果加过，可以忽略 export HADOOP_CLASSPATH=`hadoop classpath`

加载配置

$ source /etc/profile

1、yarn-session模式（Seesion Mode）

Yarn-session模式下，首先向Yarn提交一个长时运行的空应用，运行起来之后，任务跑完集群也不释放，可以重复使用在Yarn上开启的Flink集群，也称为共享型集群，适合小任务。

实验

第一种模式分为两步：yarn-session.sh(启动，开辟/申请资源)+flink run（提交任务）

【第一步】开源资源，使用如下命令：

$ yarn-session.sh -n 2 -jm 1024 -tm 1024 -d ### 参数解释： #-n 2 表示指定两个容器 # -jm 1024 表示jobmanager 1024M内存 # -tm 1024表示taskmanager 1024M内存 #-d 任务后台运行 ### 如果你不希望flink yarn client一直运行，也可以启动一个后台运行的yarn session。使用这个参数：-d 或者 --detached。在这种情况下，flink yarn client将会只提交任务到集群然后关闭自己。注意：在这种情况下，无法使用flink停止yarn session，必须使用yarn工具来停止yarn session。 # yarn application -kill $applicationId #-nm,--name YARN上为一个自定义的应用设置一个名字 #-q,--query 显示yarn中可用的资源 (内存, cpu核数) #-z,--zookeeperNamespace 针对HA模式在zookeeper上创建NameSpace #-id,--applicationId YARN集群上的任务id，附着到一个后台运行的yarn session中

【第二步】提交任务为了进行测试，我们对Flink目录下的LICENSE文件进行词频统计

$ cd $FLINK_HOME $ hadoop fs -put LICENSE / $ hadoop fs -ls /LICENSE # 提交任务 $ flink run ./examples/batch/WordCount.jar -input hdfs://hadoop-node1:8082/LICENSE -output hdfs://hadoop-node1:8082/wordcount-result.txt

再提交一次任务

【注意】-output一定是不存在的文件，有flink自动创建写入

$ flink run ./examples/batch/WordCount.jar -input hdfs://hadoop-node1:8082/LICENSE -output hdfs://hadoop-node1:8082/wordcount-result2.txt

2、yarn-cluster模式（Per-Job Mode）【推荐】

Yarn-cluster模式下，每个任务单独在Yarn上启动一套Flink集群,适合大任务，运行完后结束，集群释放，资源释放，再有任务，会再起新的Flink集群，需要频繁的在Yanr上开启Flink集群，集群相互独立，适合大任务。

当然除了on yarn模式，还有on k8s，有兴趣的小伙伴，可以试试，当时目前企业里用的最多的还是on yarn模式，但是现在不是流行容器化嘛，以后很大可能会慢慢转到 on k8s模式。

实验

第二种模式其实也分为两个部分，依然是开辟资源和提交任务，但是在Job模式下，这两步都合成一个命令了。

$ cd $FLINK_HOME $ flink run -m yarn-cluster -yjm 1024 -ytm 1024 ./examples/batch/WordCount.jar # 查看帮助 $ flink --help ### 参数详解，这里只列出了部分参数 Options for yarn-cluster mode: -d,--detached If present, runs the job in detached mode -m,--jobmanager Set to yarn-cluster to use YARN execution mode. -yat,--yarnapplicationType Set a custom application type for the application on YARN -yD use value for given property -yd,--yarndetached If present, runs the job in detached mode (deprecated; use non-YARN specific option instead) -yh,--yarnhelp Help for the Yarn session CLI. -yid,--yarnapplicationId Attach to running YARN session -yj,--yarnjar Path to Flink jar file -yjm,--yarnjobManagerMemory Memory for JobManager Container with optional unit (default: MB) -ynl,--yarnnodeLabel Specify YARN node label for the YARN application -ynm,--yarnname Set a custom name for the application on YARN -yq,--yarnquery Display available YARN resources (memory, cores) -yqu,--yarnqueue Specify YARN queue. -ys,--yarnslots Number of slots per TaskManager -yt,--yarnship Ship files in the specified directory (t for transfer) -ytm,--yarntaskManagerMemory Memory per TaskManager Container with optional unit (default: MB) -yz,--yarnzookeeperNamespace Namespace to create the Zookeeper sub-paths for high availability mode -z,--zookeeperNamespace Namespace to create the Zookeeper sub-paths for high availability mode

【温馨提示】上面命令中没有指定-input 和 -output，这是由于有默认的数据集和输出方式，看看效果。

发现查看不了History，是因为没起History服务，下面启动这个服务

historyserver简介与配置

History Server允许查询由JobManager归档的已完成作业的状态和统计信息。已完成作业的归档在JobManager上进行，JobManager会将归档的作业信息upload到文件系统目录，这个文件系统可以是本地文件系统、HDFS、H3等，这个目录是可以在配置文件中指定的。然后还需要配置History Server去扫描这个目录，并且可以配置扫描的间隔时间。

配置historyserver

$ cd $FLINK_HOME/bin # 选创建目录 $ hdfs://hadoop-node1:8082/flink/completed-jobs/ # conf/flink-conf.yaml # 指定由JobManager归档的作业信息所存放的目录，这里使用的是HDFS jobmanager.archive.fs.dir: hdfs://hadoop-node1:8082/flink/completed-jobs/ # 指定History Server扫描哪些归档目录，多个目录使用逗号分隔 historyserver.archive.fs.dir: hdfs://hadoop-node1:8082/flink/completed-jobs/ # 指定History Server间隔多少毫秒扫描一次归档目录 historyserver.archive.fs.refresh-interval: 10000 # History Server所绑定的ip，hadoop-node1代表允许所有ip访问 historyserver.web.address: hadoop-node1 # 指定History Server所监听的端口号，默认端口是8082 historyserver.web.port: 9082

启动historyserver

$ ./historyserver.sh start $ jps

web：flink run -m yarn-cluster -yjm 1024 -ytm 1024 ./examples/batch/WordCount.jar

3、Application模式（Application Mode）

Application模式的由来

其实前面两种模式client端还是需要干三件事情的：

获取作业所需的依赖项； 通过执行环境分析并取得逻辑计划，即StreamGraph→JobGraph； 将依赖项和JobGraph上传到集群中。 只有在这些都完成之后，才会通过env.execute()方法触发Flink运行时真正地开始执行作业。试想，如果所有用户都在Deployer上提交作业，较大的依赖会消耗更多的带宽，而较复杂的作业逻辑翻译成JobGraph也需要吃掉更多的CPU和内存，客户端的资源反而会成为瓶颈——不管Session还是Per-Job模式都存在此问题。为了解决它，社区在传统部署模式的基础上实现了Application模式。

Application模式概述

Application模式原本需要客户端做的三件事被转移到了JobManager里，也就是说main()方法在集群中执行（入口点位于ApplicationClusterEntryPoint），Deployer只需要负责发起部署请求了。另外，如果一个main()方法中有多个env.execute()/executeAsync()调用，在Application模式下，这些作业会被视为属于同一个应用，在同一个集群中执行（如果在Per-Job模式下，就会启动多个集群）。可见，Application模式本质上是Session和Per-Job模式的折衷。

实验

$ cd $FLINK_HOME $ ./bin/flink run-application -t yarn-application \ -Djobmanager.memory.process.size=2048m \ -Dtaskmanager.memory.process.size=4096m \ -Dtaskmanager.numberOfTaskSlots=2 \ -Dparallelism.default=10 \ -Dyarn.application.name="MyFlinkApp" \ ./examples/batch/WordCount.jar

【温馨提示】 -t参数用来指定部署目标，目前支持YARN（yarn-application）和K8S（kubernetes-application）。-D参数则用来指定与作业相关的各项参数，具体可参见官方文档。 六、Spark与Flink对比

Flink原理介绍先写到这里了，更多关于Flink的知识点，请您耐心等待，当然也可以先自行去看官方文档：https://nightlies.apache.org/flink/。

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。