实时大数据计算引擎Apache Flink计算研究(二)

接上文 实时大数据计算引擎Apache Flink计算研究(一)

8、Flink DataSetAPI

数据源部分

不但提供了流处理 还提供了批处理 流处理和批处理分属于不同的api

基于文件

readTextFile(path)

基于集合

fromCollection(Collection)

算子部分

Map 输入一个元素 然后返回一个元素 中间可以做一些清洗转换等操作

FlatMap 输入一个元素 可以返回零个 一个或者多个元素

MapPartition 类似map 一次处理一个分区的数据【如果在进行map处理的时候需要获取第三方资源链接 建议使用MapPartition】

Filter 过滤函数 对传入的数据进行判断 符合条件的数据会被留下

Reduce 对数据进行聚合操作 结合当前元素和上一次reduce返回的值进行聚合操作 然后返回一个新的值

Aggregate sum、max、min等

Distinct 返回一个数据集中去重之后的元素 data.distinct()

Join 内连接

OuterJoin 外链接

Cross 获取两个数据集的笛卡尔积

Union 返回两个数据集的总和 数据类型需要一致

First-n 获取集合中的前N个元素

Sort Partition 在本地对数据集的所有分区进行排序 通过sortPartition()的链接调用来完成对多个字段的排序

批处理的统计文件中单词的个数

运行结果

输入文件

输出文件

9、DataSet API之Transformations

获取笛卡尔积

运行结果

批处理去重

处理结果

获取集合中的前N个元素

运行结果

10、DataSet API之Data sinks

writeAsText() 将元素以字符串形式逐行写入 这些字符串通过调用每个元素的toString()方法来获取

writeAsCsv() 将元组以逗号分隔写入文件中 行及字段之间的分隔是可配置的。每个字段的值来自对象的toString()方法

print() 打印每个元素的toString()方法的值到标准输出或者标准错误输出流中

在前2个章节中已测试过其中的方法。

11、Flink中的广播变量

1.应用场景

把元素广播给所有的分区 数据会被重复处理

广播变量允许编程人员在每台机器上保持1个只读的缓存变量 而不是传送变量的副本给tasks

广播变量创建后 它可以运行在集群中的任何function上 而不需要多次传递给集群节点。另外需要记住 不应该修改广播变量 这样才能确保每个节点获取到的值都是一致的

一句话解释 可以理解为是一个公共的共享变量 我们可以把一个dataset 数据集广播出去 然后不同的task在节点上都能够获取到 这个数据在每个节点上只会存在一份。如果不使用broadcast 则在每个节点中的每个task中都需要拷贝一份dataset数据集 比较浪费内存(也就是一个节点中可能会存在多份dataset数据)。

用法

1 初始化数据DataSet toBroadcast env.fromElements(1, 2, 3)2 广播数据.withBroadcastSet(toBroadcast, broadcastSetName 3 获取数据Collection broadcastSet getRuntimeContext().getBroadcastVariable( broadcastSetName

注意

1 广播出去的变量存在于每个节点的内存中 所以这个数据集不能太大。因为广播出去的数据 会常驻内存 除非程序执行结束2 广播变量在初始化广播出去以后不支持修改 这样才能保证每个节点的数据都是一致的。

2.代码实现

12、Flink的累加器

1.Accumulator即累加器 与Mapreduce counter的应用场景差不多 都能很好地观察task在运行期间的数据变化

可以在Flink job任务中的算子函数中操作累加器 但是只能在任务执行结束之后才能获得累加器的最终结果。

Counter是一个具体的累加器(Accumulator)实现

IntCounter, LongCounter 和 DoubleCounter

l用法

1 创建累加器private IntCounter numLines new IntCounter();2 注册累加器getRuntimeContext().addAccumulator( num-lines , this.numLines);3 使用累加器this.numLines.add(1);4 获取累加器的结果

myJobExecutionResult.getAccumulatorResult( num-lines )

2.代码

运行结果

13、Flink Distributed Cache(分布式缓存)

Flink提供了一个分布式缓存 类似于hadoop 可以使用户在并行函数中很方便的读取本地文件

此缓存的工作机制如下 程序注册一个文件或者目录(本地或者远程文件系统) 通过ExecutionEnvironment注册缓存文件并为它起一个名称。当程序执行 Flink自动将文件或者目录复制到所有taskmanager节点的本地文件系统 用户可以通过这个指定的名称查找文件或者目录 然后从taskmanager节点的本地文件系统访问它

用法

1 注册一个文件env.registerCachedFile( hdfs:///path/to/your/file , hdfsFile )2 访问数据File myFile getRuntimeContext().getDistributedCache().getFile( hdfsFile

代码

运行结果

14、状态(State)

我们前面写的word count的例子 没有包含状态管理。如果一个task在处理过程中挂掉了 那么它在内存中的状态都会丢失 所有的数据都需要重新计算。从容错和消息处理的语义上(at least once, exactly once) Flink引入了state和checkpoint。

首先区分一下两个概念?

state一般指一个具体的task/operator的状态【state数据默认保存在java的堆内存中】而checkpoint【可以理解为checkpoint是把state数据持久化存储了】 则表示了一个Flink Job在一个特定时刻的一份全局状态快照 即包含了所有task/operator的状态注意 task是Flink中执行的基本单位。operator指算子(transformation)。

State可以被记录 在失败的情况下数据还可以恢复

Flink中有两种基本类型的State

Keyed State

Operator State

Keyed State和Operator State 可以以两种形式存在

原始状态(raw state)托管状态(managed state)

托管状态是由Flink框架管理的状态

而原始状态 由用户自行管理状态具体的数据结构 框架在做checkpoint的时候 使用byte[]来读写状态内容 对其内部数据结构一无所知。

1.State-Keyed State介绍

顾名思义 就是基于KeyedStream上的状态。这个状态是跟特定的key绑定的 对KeyedStream流上的每一个key 都对应一个state。

stream.keyBy(…)

保存state的数据结构

ValueState:即类型为T的单值状态。这个状态与对应的key绑定 是最简单的状态了。它可以通过update方法更新状态值 通过value()方法获取状态值ListState:即key上的状态值为一个列表。可以通过add方法往列表中附加值 也可以通过get()方法返回一个Iterable来遍历状态值ReducingState:这种状态通过用户传入的reduceFunction 每次调用add方法添加值的时候 会调用reduceFunction 最后合并到一个单一的状态值MapState:即状态值为一个map。用户通过put或putAll方法添加元素

需要注意的是 以上所述的State对象 仅仅用于与状态进行交互 更新、删除、清空等 而真正的状态值 有可能是存在内存、磁盘、或者其他分布式存储系统中。相当于我们只是持有了这个状态的句柄

2.State-Operator State

与Key无关的State 与Operator绑定的state 整个operator只对应一个state

保存state的数据结构

ListState

举例来说 Flink中的Kafka Connector 就使用了operator state。它会在每个connector实例中 保存该实例中消费topic的所有(partition, offset)映射

checkPoint简介

为了保证state的容错性 Flink需要对state进行checkpoint。

Checkpoint是Flink实现容错机制最核心的功能 它能够根据配置周期性地基于Stream中各个Operator/task的状态来生成快照 从而将这些状态数据定期持久化存储下来 当Flink程序一旦意外崩溃时 重新运行程序时可以有选择地从这些快照进行恢复 从而修正因为故障带来的程序数据异常

Flink的checkpoint机制可以与(stream和state)的持久化存储交互的前提

持久化的source 它需要支持在一定时间内重放事件。这种sources的典型例子是持久化的消息队列 比如Apache Kafka RabbitMQ等 或文件系统 比如HDFS S3 GFS等 用于state的持久化存储 例如分布式文件系统 比如HDFS S3 GFS等

默认checkpoint功能是disabled的 想要使用的时候需要先启用

checkpoint开启之后 默认的checkPointMode是Exactly-once

checkpoint的checkPointMode有两种 Exactly-once和At-least-once

Exactly-once对于大多数应用来说是最合适的。At-least-once可能用在某些延迟超低的应用程序 始终延迟为几毫秒

默认checkpoint功能是disabled的 想要使用的时候需要先启用

StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();

// 每隔1000 ms进行启动一个检查点【设置checkpoint的周期】

env.enableCheckpointing(1000);

// 高级选项

// 设置模式为exactly-once 这是默认值

env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// 确保检查点之间有至少500 ms的间隔【checkpoint最小间隔】

env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);

// 检查点必须在一分钟内完成 或者被丢弃【checkpoint的超时时间】

env.getCheckpointConfig().setCheckpointTimeout(60000);

// 同一时间只允许进行一个检查点

env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

// 表示一旦Flink处理程序被cancel后 会保留Checkpoint数据 以便根据实际需要恢复到指定的Checkpoint【详细解释见备注】

env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

State Backend(状态的后端存储)

默认情况下 state会保存在taskmanager的内存中 checkpoint会存储在JobManager的内存中。

state 的store和checkpoint的位置取决于State Backend的配置

env.setStateBackend(…)

一共有三种State Backend

MemoryStateBackendFsStateBackend

RocksDBStateBackend

MemoryStateBackend

state数据保存在java堆内存中 执行checkpoint的时候 会把state的快照数据保存到jobmanager的内存中基于内存的state backend在生产环境下不建议使用

FsStateBackend

state数据保存在taskmanager的内存中 执行checkpoint的时候 会把state的快照数据保存到配置的文件系统中可以使用hdfs等分布式文件系统

RocksDBStateBackend

RocksDB跟上面的都略有不同 它会在本地文件系统中维护状态 state会直接写入本地rocksdb中。同时它需要配置一个远端的filesystem uri 一般是HDFS 在做checkpoint的时候 会把本地的数据直接复制到filesystem中。fail over的时候从filesystem中恢复到本地RocksDB克服了state受内存限制的缺点 同时又能够持久化到远端文件系统中 比较适合在生产中使用l修改State Backend的两种方式第一种 单任务调整修改当前任务代码env.setStateBackend(new FsStateBackend( hdfs://namenode:9000/flink/checkpoints 或者new MemoryStateBackend()或者new RocksDBStateBackend(filebackend, true);【需要添加第三方依赖】第二种 全局调整
修改flink-conf.yamlstate.backend: filesystemstate.checkpoints.dir: hdfs://namenode:9000/flink/checkpoints注意 state.backend的值可以是下面几种 jobmanager(MemoryStateBackend), filesystem(FsStateBackend), rocksdb(RocksDBStateBackend)Restart Strategies(重启策略)

Flink支持不同的重启策略 以在故障发生时控制作业如何重启

集群在启动时会伴随一个默认的重启策略 在没有定义具体重启策略时会使用该默认策略。 如果在工作提交时指定了一个重启策略 该策略会覆盖集群的默认策略

默认的重启策略可以通过 Flink 的配置文件 flink-conf.yaml 指定。配置参数 restart-strategy 定义了哪个策略被使用。

常用的重启策略

固定间隔 (Fixed delay)失败率 (Failure rate)无重启 (No restart)

如果没有启用 checkpointing 则使用无重启 (no restart) 策略。

如果启用了 checkpointing 但没有配置重启策略 则使用固定间隔 (fixed-delay) 策略 其中 Integer.MAX_VALUE 参数是尝试重启次数

重启策略可以在flink-conf.yaml中配置 表示全局的配置。也可以在应用代码中动态指定 会覆盖全局配置

重启策略之固定间隔 (Fixed delay)

l第一种 全局配置 flink-conf.yaml

restart-strategy: fixed-delayrestart-strategy.fixed-delay.attempts: 3restart-strategy.fixed-delay.delay: 10 s

l第二种 应用代码设置

env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, // 尝试重启的次数Time.of(10, TimeUnit.SECONDS) // 间隔));

重启策略之失败率 (Failure rate)

第一种 全局配置 flink-conf.yaml

restart-strategy: failure-raterestart-strategy.failure-rate.max-failures-per-interval: 3restart-strategy.failure-rate.failure-rate-interval: 5 minrestart-strategy.failure-rate.delay: 10 s

第二种 应用代码设置

env.setRestartStrategy(RestartStrategies.failureRateRestart(3, // 一个时间段内的最大失败次数Time.of(5, TimeUnit.MINUTES), // 衡量失败次数的是时间段Time.of(10, TimeUnit.SECONDS) // 间隔

));

重启策略之无重启 (No restart)

第一种 全局配置 flink-conf.yaml

restart-strategy: none

第二种 应用代码设置

env.setRestartStrategy(RestartStrategies.noRestart());

保存多个Checkpoint

默认情况下 如果设置了Checkpoint选项 则Flink只保留最近成功生成的1个Checkpoint 而当Flink程序失败时 可以从最近的这个Checkpoint来进行恢复。但是 如果我们希望保留多个Checkpoint 并能够根据实际需要选择其中一个进行恢复 这样会更加灵活 比如 我们发现最近4个小时数据记录处理有问题 希望将整个状态还原到4小时之前

Flink可以支持保留多个Checkpoint 需要在Flink的配置文件conf/flink-conf.yaml中 添加如下配置 指定最多需要保存Checkpoint的个数

state.checkpoints.num-retained: 20

这样设置以后就查看对应的Checkpoint在HDFS上存储的文件目录

hdfs dfs -ls hdfs://namenode:9000/flink/checkpoints如果希望回退到某个Checkpoint点 只需要指定对应的某个Checkpoint路径即可实现

l如果Flink程序异常失败 或者最近一段时间内数据处理错误 我们可以将程序从某一个Checkpoint点进行恢复

lbin/flink run -s hdfs://namenode:9000/flink/checkpoints/467e17d2cc343e6c56255d222bae3421/chk-56/_metadata flink-job.jar

l程序正常运行后 还会按照Checkpoint配置进行运行 继续生成Checkpoint数据

savePoint

Flink通过Savepoint功能可以做到程序升级后 继续从升级前的那个点开始执行计算 保证数据不中断

全局 一致性快照。可以保存数据源offset operator操作状态等信息

可以从应用在过去任意做了savepoint的时刻开始继续消费

checkPoint?vs?savePoint

checkPoint

应用定时触发 用于保存状态 会过期内部应用失败重启的时候使用

savePoint

用户手动执行 是指向Checkpoint的指针 不会过期在升级的情况下使用注意 为了能够在作业的不同版本之间以及 Flink 的不同版本之间顺利升级 强烈推荐程序员通过 uid(String) 方法手动的给算子赋予 ID 这些 ID 将用于确定每一个算子的状态范围。如果不手动给各算子指定 ID 则会由 Flink 自动给每个算子生成一个 ID。只要这些 ID 没有改变就能从保存点 savepoint 将程序恢复回来。而这些自动生成的 ID 依赖于程序的结构 并且对代码的更改是很敏感的。因此 强烈建议用户手动的设置 ID。

savePoint的使用

1 在flink-conf.yaml中配置Savepoint存储位置

不是必须设置 但是设置后 后面创建指定Job的Savepoint时 可以不用在手动执行命令时指定Savepoint的位置state.savepoints.dir: hdfs://namenode:9000/flink/savepoints

2 触发一个savepoint【直接触发或者在cancel的时候触发】

bin/flink savepoint jobId [targetDirectory] [-yid yarnAppId]【针对on yarn模式需要指定-yid参数】bin/flink cancel -s [targetDirectory] jobId [-yid yarnAppId]【针对on yarn模式需要指定-yid参数】

3 从指定的savepoint启动job

bin/flink run -s savepointPath [runArgs]

THE END

本文转自网络，原文链接：https://developer.aliyun.com/article/784959
本站部分内容转载于网络，版权归原作者所有，转载之目的在于传播更多优秀技术内容，如有侵权请联系QQ/微信：153890879删除，谢谢！