4) Flink 关联 Hive 分区表

Flink 1．12 支持了 Hive 最新的分区作为时态表的功能,可以通过 SQL 的方式直接关联 Hive 分区表的最新分区,并且会自动监听最新的 Hive 分区,当监控到新的分区后,会自动地做维表数据的全量替换。通过这种方式,用户无需编写 DataStream 程序即可完成 Kafka 流实时关联最新的 Hive 分区实现数据打宽。

具体用法:

在 Sql Client 中注册 HiveCatalog:

vim conf/sql-client-defaults．yaml
catalogs:
 - name: hive_catalog
   type: hive
   hive-conf-dir: /disk0/soft/hive-conf/ #该目录需要包hive-site．xml文件

创建 Kafka 表

CREATE TABLE hive_catalog．flink_db．kfk_fact_bill_master_12 (  
   master Row

Flink 事实表与 Hive 最新分区数据关联

dim_extend_shop_info 是 Hive 中已存在的表,所以我们用 table hint 动态地开启维表参数。

CREATE VIEW IF NOT EXISTS hive_catalog．flink_db．view_fact_bill_master as  
SELECT * FROM  
(select t1．*, t2．group_id, t2．shop_id, t2．group_name, t2．shop_name, t2．brand_id,  
    ROW_NUMBER() OVER (PARTITION BY groupID, shopID, orderKey ORDER BY actionTime desc) rn  
   from hive_catalog．flink_db．kfk_fact_bill_master_12 t1  
      JOIN hive_catalog．flink_db．dim_extend_shop_info  
 + OPTIONS('streaming-source．enable'='true',  
    'streaming-source．partition．include' = 'latest',  
    'streaming-source．monitor-interval' = '1 h',
    'streaming-source．partition-order' = 'partition-name')
   FOR SYSTEM_TIME AS OF t1．proctime AS t2 --时态表  
   ON t1．groupID = t2．group_id and t1．shopID = t2．shop_id  
   where groupID in (202042)) t  where t．rn = 1

参数解释:

streaming-source．enable 开启流式读取 Hive 数据。

streaming-source．partition．include 有以下两个值:

latest 属性: 只读取最新分区数据。all: 读取全量分区数据 ,默认值为 all,表示读所有分区,latest 只能用在 temporal join 中,用于读取最新分区作为维表,不能直接读取最新分区数据。

streaming-source．monitor-interval 监听新分区生成的时间、不宜过短 、最短是1 个小时,因为目前的实现是每个 task 都会查询 metastore,高频的查可能会对metastore 产生过大的压力。需要注意的是,1．12．1 放开了这个限制,但仍建议按照实际业务不要配个太短的 interval。

streaming-source．partition-order 分区策略,主要有以下 3 种,其中最为推荐的是 partition-name:

partition-name 使用默认分区名称顺序加载最新分区create-time 使用分区文件创建时间顺序partition-time 使用分区时间顺序六、Flink 状态管理

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

因此可以说flink因为引入了state和checkpoint所以才支持的exactly once

首先区分一下两个概念:

state:

state一般指一个具体的task/operator的状态:

state数据默认保存在java的堆内存中,TaskManage节点的内存中。

operator表示一些算子在运行的过程中会产生的一些中间结果。

checkpoint:

checkpoint可以理解为checkpoint是把state数据定时持久化存储了,则表示了一个Flink Job在一个特定时刻的一份全局状态快照,即包含了所有task/operator的状态。

注意:task(subTask)是Flink中执行的基本单位。operator指算子(transformation)

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

Flink中有两种基本类型的State:

Keyed State

Operator State

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

原始状态(raw state)

托管状态(managed state)

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

我们说operator算子保存了数据的中间结果,中间结果保存在什么类型中,如果我们这里是托管状态,则由flink框架自行管理

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

通常在DataStream上的状态推荐使用托管的状态,当实现一个用户自定义的operator时,会使用到原始状态。

1． State-Keyed State

基于KeyedStream上的状态。这个状态是跟特定的key绑定的,对KeyedStream流上的每一个key,都对应一个state,比如:stream．keyBy(…)。KeyBy之后的Operator State,可以理解为分区过的Operator State。

保存state的数据结构:

ValueState:即类型为T的单值状态。这个状态与对应的key绑定,是最简单的状态了。它可以通过update方法更新状态值,通过value()方法获取状态值。

ListState:即key上的状态值为一个列表。可以通过add方法往列表中附加值;也可以通过get()方法返回一个Iterable来遍历状态值。

ReducingState:这种状态通过用户传入的reduceFunction,每次调用add方法添加值的时候,会调用reduceFunction,最后合并到一个单一的状态值。

MapState

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

1． ValueState

使用ValueState保存中间结果对下面数据进行分组求和。

开发步骤:

1． 获取流处理执行环境
 2． 加载数据源
 3． 数据分组
 4． 数据转换,定义ValueState,保存中间结果
 5． 数据打印
 6． 触发执行

ValueState:测试数据源:

List(
  (1L, 4L),
  (2L, 3L),
  (3L, 1L),
  (1L, 2L),
  (3L, 2L),
  (1L, 2L),
  (2L, 2L),
  (2L, 9L)
)

示例代码:

import org．apache．flink．api．common．functions．RichFlatMapFunction
import org．apache．flink．api．common．state．{ValueState, ValueStateDescriptor}
import org．apache．flink．api．common．typeinfo．{TypeHint, TypeInformation}
import org．apache．flink．configuration．Configuration
import org．apache．flink．streaming．api．scala．_
import org．apache．flink．streaming．api．scala．{DataStream, StreamExecutionEnvironment}
import org．apache．flink．util．Collector
object TestKeyedState {
 class CountWithKeyedState extends RichFlatMapFunction[(Long, Long), (Long, Long)] {
   *
    * ValueState状态句柄． 第一个值为count,第二个值为sum。
   
   private var sum: ValueState[(Long, Long)] = _
   override def flatMap(input: (Long, Long), out: Collector[(Long, Long)]): Unit = {
     // 获取当前状态值
     val tmpCurrentSum: (Long, Long) = sum．value
     // 状态默认值
     val currentSum = if (tmpCurrentSum != null) {
       tmpCurrentSum
     } else {
       (0L, 0L)
     }
     // 更新
     val newSum = (currentSum．_1 + 1, currentSum．_2 + input．_2)
     // 更新状态值
     sum．update(newSum)
     // 如果count >=3 清空状态值,重新计算
     if (newSum．_1 >= 3) {
       out．collect((input．_1, newSum．_2 / newSum．_1))
       sum．clear()
     }
   }
   override def open(parameters: Configuration): Unit = {
     sum = getRuntimeContext．getState(
       new ValueStateDescriptor[(Long, Long)]("average", // 状态名称
         TypeInformation．of(new TypeHint[(Long, Long)](){}) )// 状态类型
     )
   }
 }  
 def main(args: Array[String]): Unit = {
   //初始化执行环境
   val env = StreamExecutionEnvironment．getExecutionEnvironment
   //构建数据源
   val inputStream: DataStream[(Long, Long)] = env．fromCollection(
     List(
       (1L, 4L),
       (2L, 3L),
       (3L, 1L),
       (1L, 2L),
       (3L, 2L),
       (1L, 2L),
       (2L, 2L),
       (2L, 9L))
   )
   //执行数据处理
   inputStream．keyBy(0)
     ．flatMap(new CountWithKeyedState)
     ．setParallelism(1)
     ．print
   //运行任务
   env．execute
 }
}  
2． MapState

使用MapState保存中间结果对下面数据进行分组求和:

1． 获取流处理执行环境
 2． 加载数据源
 3． 数据分组
 4． 数据转换,定义MapState,保存中间结果
 5． 数据打印
 6． 触发执行

MapState:测试数据源:

List(
  ("java", 1),
  ("python", 3),
  ("java", 2),
  ("scala", 2),
  ("python", 1),
  ("java", 1),
  ("scala", 2)
)  

示例代码:

object MapState {
 def main(args: Array[String]): Unit = {
   val env: StreamExecutionEnvironment = StreamExecutionEnvironment．getExecutionEnvironment
   env．setParallelism(1)
   *
     * 使用MapState保存中间结果对下面数据进行分组求和
     * 1．获取流处理执行环境
     * 2．加载数据源
     * 3．数据分组
     * 4．数据转换,定义MapState,保存中间结果
     * 5．数据打印
     * 6．触发执行
     
   val source: DataStream[(String, Int)] = env．fromCollection(List(
     ("java", 1),
     ("python", 3),
     ("java", 2),
     ("scala", 2),
     ("python", 1),
     ("java", 1),
     ("scala", 2)))
 
   source．keyBy(0)
     ．map(new RichMapFunction[(String, Int), (String, Int)] {
       var mste: MapState[String, Int] = _
       override def open(parameters: Configuration): Unit = {
         val msState = new MapStateDescriptor[String, Int]("ms",
           TypeInformation．of(new TypeHint[(String)] {}),
           TypeInformation．of(new TypeHint[(Int)] {}))
         mste = getRuntimeContext．getMapState(msState)
       }
       override def map(value: (String, Int)): (String, Int) = {
         val i: Int = mste．get(value．_1)
         mste．put(value．_1, value．_2 + i)
         (value．_1, value．_2 + i)
       }
     })．print()
   env．execute()
 }
}  
2． State-Operator State

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

保存state的数据结构:

ListState

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

步骤:

获取执行环境

设置检查点机制:路径,重启策略

自定义数据源

需要继承并行数据源和CheckpointedFunction设置listState,通过上下文对象context获取数据处理,保留offset制作快照

数据打印

触发执行

示例代码:

import java．util
import org．apache．flink．api．common．restartstrategy．RestartStrategies
import org．apache．flink．api．common．state．{ListState, ListStateDescriptor}
import org．apache．flink．api．common．time．Time
import org．apache．flink．api．common．typeinfo．{TypeHint, TypeInformation}
import org．apache．flink．runtime．state．{FunctionInitializationContext, FunctionSnapshotContext}
import org．apache．flink．runtime．state．filesystem．FsStateBackend
import org．apache．flink．streaming．api．CheckpointingMode
import org．apache．flink．streaming．api．checkpoint．CheckpointedFunction
import org．apache．flink．streaming．api．environment．CheckpointConfig
import org．apache．flink．streaming．api．functions．source．{RichParallelSourceFunction, SourceFunction}
import org．apache．flink．streaming．api．scala．StreamExecutionEnvironment
import org．apache．flink．streaming．api．scala．_
object ListOperate {
 def main(args: Array[String]): Unit = {
   val env: StreamExecutionEnvironment = StreamExecutionEnvironment．getExecutionEnvironment
   env．setParallelism(1)
   env．enableCheckpointing(5000)
   env．setStateBackend(new FsStateBackend("hdfs://node01:8020/tmp/check/8"))
   env．getCheckpointConfig．setCheckpointingMode(CheckpointingMode．EXACTLY_ONCE)
   env．getCheckpointConfig．setMaxConcurrentCheckpoints(1)
   env．getCheckpointConfig．setCheckpointTimeout(60000)
   env．getCheckpointConfig．setFailOnCheckpointingErrors(false)
   env．getCheckpointConfig．enableExternalizedCheckpoints(CheckpointConfig．ExternalizedCheckpointCleanup．RETAIN_ON_CANCELLATION)
   //重启策略
   env．setRestartStrategy(RestartStrategies．failureRateRestart(3, Time．minutes(1), Time．seconds(5)))
   //模拟kakfa偏移量
   env．addSource(new MyRichParrelSourceFun)
     ．print()
   env．execute()
 }
}
class MyRichParrelSourceFun extends RichParallelSourceFunction[String]
 with CheckpointedFunction {
 var listState: ListState[Long] = _
 var offset: Long = 0L
 //任务运行
 override def run(ctx: SourceFunction．SourceContext[String]): Unit = {
   val iterState: util．Iterator[Long] = listState．get()．iterator()
   while (iterState．hasNext) {
     offset = iterState．next()
   }
   while (true) {
     offset += 1
     ctx．collect("offset:"+offset)
     Thread．sleep(1000)
     if(offset > 10){
       1/0
     }
   }
 }
 //取消任务
 override def cancel(): Unit = ???
 //制作快照
 override def snapshotState(context: FunctionSnapshotContext): Unit = {
   listState．clear()
   listState．add(offset)
 }
 //初始化状态
 override def initializeState(context: FunctionInitializationContext): Unit = {
   listState = context．getOperatorStateStore．getListState(new ListStateDescriptor[Long](
     "listState", TypeInformation．of(new TypeHint[Long] {})
   ))
 }
}
3． Broadcast State

Broadcast State 是 Flink 1．5 引入的新特性。在开发过程中,如果遇到需要下发/广播配置、规则等低吞吐事件流到下游所有 task 时,就可以使用 Broadcast State 特性。下游的 task 接收这些配置、规则并保存为 BroadcastState, 将这些配置应用到另一个数据流的计算中 。

1) API介绍

通常,我们首先会创建一个Keyed或Non-Keyed的Data Stream,然后再创建一个Broadcasted Stream,最后通过Data Stream来连接(调用connect方法)到Broadcasted Stream上,这样实现将Broadcast State广播到Data Stream下游的每个Task中。

如果Data Stream是Keyed Stream,则连接到Broadcasted Stream后,添加处理ProcessFunction时需要使用KeyedBroadcastProcessFunction来实现,下面是KeyedBroadcastProcessFunction的API,代码如下所示:

public abstract class KeyedBroadcastProcessFunction

上面泛型中的各个参数的含义,说明如下:

KS:表示Flink程序从最上游的Source Operator开始构建Stream,当调用keyBy时所依赖的Key的类型;IN1:表示非Broadcast的Data Stream中的数据记录的类型;IN2:表示Broadcast Stream中的数据记录的类型;OUT:表示经过KeyedBroadcastProcessFunction的processElement()和processBroadcastElement()方法处理后输出结果数据记录的类型。

如果Data Stream是Non-Keyed Stream,则连接到Broadcasted Stream后,添加处理ProcessFunction时需要使用BroadcastProcessFunction来实现,下面是BroadcastProcessFunction的API,代码如下所示:

public abstract class BroadcastProcessFunction

上面泛型中的各个参数的含义,与前面KeyedBroadcastProcessFunction的泛型类型中的后3个含义相同,只是没有调用keyBy操作对原始Stream进行分区操作,就不需要KS泛型参数。

注意事项:

Broadcast State 是Map类型,即K-V类型。

Broadcast State 只有在广播一侧的方法中processBroadcastElement可以修改;在非广播一侧方法中processElement只读。

Broadcast State在运行时保存在内存中。

2) 场景举例

动态更新计算规则: 如事件流需要根据最新的规则进行计算,则可将规则作为广播状态广播到下游Task中。

实时增加额外字段: 如事件流需要实时增加用户的基础信息,则可将用户的基础信息作为广播状态广播到下游Task中。

七、Flink的容错1． Checkpoint介绍

checkpoint机制是Flink可靠性的基石,可以保证Flink集群在某个算子因为某些原因(如 异常退出)出现故障时,能够将整个应用流图的状态恢复到故障之前的某一状态,保 证应用流图状态的一致性。Flink的checkpoint机制原理来自“Chandy-Lamport algorithm”算法。

每个需要checkpoint的应用在启动时,Flink的JobManager为其创建一个 CheckpointCoordinator(检查点协调器),CheckpointCoordinator全权负责本应用的快照制作。

CheckpointCoordinator(检查点协调器) 周期性的向该流应用的所有source算子发送 barrier(屏障)。

当某个source算子收到一个barrier时,便暂停数据处理过程,然后将自己的当前状态制作成快照,并保存到指定的持久化存储中,最后向CheckpointCoordinator报告自己快照制作情况,同时向自身所有下游算子广播该barrier,恢复数据处理

下游算子收到barrier之后,会暂停自己的数据处理过程,然后将自身的相关状态制作成快照,并保存到指定的持久化存储中,最后向CheckpointCoordinator报告自身快照情况,同时向自身所有下游算子广播该barrier,恢复数据处理。

每个算子按照步骤3不断制作快照并向下游广播,直到最后barrier传递到sink算子,快照制作完成。

当CheckpointCoordinator收到所有算子的报告之后,认为该周期的快照制作成功; 否则,如果在规定的时间内没有收到所有算子的报告,则认为本周期快照制作失败。

如果一个算子有两个输入源,则暂时阻塞先收到barrier的输入源,等到第二个输入源相 同编号的barrier到来时,再制作自身快照并向下游广播该barrier。具体如下图所示:

假设算子C有A和B两个输入源

在第i个快照周期中,由于某些原因(如处理时延、网络时延等)输入源A发出的 barrier 先到来,这时算子C暂时将输入源A的输入通道阻塞,仅收输入源B的数据。

当输入源B发出的barrier到来时,算子C制作自身快照并向 CheckpointCoordinator 报告自身的快照制作情况,然后将两个barrier合并为一个,向下游所有的算子广播。

当由于某些原因出现故障时,CheckpointCoordinator通知流图上所有算子统一恢复到某个周期的checkpoint状态,然后恢复数据流处理。分布式checkpoint机制保证了数据仅被处理一次(Exactly Once)。

2． 持久化存储1) MemStateBackend

该持久化存储主要将快照数据保存到JobManager的内存中,仅适合作为测试以及快照的数据量非常小时使用,并不推荐用作大规模商业部署。

MemoryStateBackend 的局限性:

默认情况下,每个状态的大小限制为 5 MB。可以在MemoryStateBackend的构造函数中增加此值。

无论配置的最大状态大小如何,状态都不能大于akka帧的大小(请参阅配置)。

聚合状态必须适合 JobManager 内存。

建议MemoryStateBackend 用于:

本地开发和调试。

状态很少的作业,例如仅包含一次记录功能的作业(Map,FlatMap,Filter,．．．),kafka的消费者需要很少的状态。

2) FsStateBackend

该持久化存储主要将快照数据保存到文件系统中,目前支持的文件系统主要是 HDFS和本地文件。如果使用HDFS,则初始化FsStateBackend时,需要传入以 “hdfs://”开头的路径(即: new FsStateBackend("hdfs:///hacluster/checkpoint")), 如果使用本地文件,则需要传入以“file://”开头的路径(即:new FsStateBackend("file:///Data"))。在分布式情况下,不推荐使用本地文件。如果某 个算子在节点A上失败,在节点B上恢复,使用本地文件时,在B上无法读取节点 A上的数据,导致状态恢复失败。

建议FsStateBackend:

具有大状态,长窗口,大键 / 值状态的作业。

所有高可用性设置。

3) RocksDBStateBackend

RocksDBStatBackend介于本地文件和HDFS之间,平时使用RocksDB的功能,将数 据持久化到本地文件中,当制作快照时,将本地数据制作成快照,并持久化到 FsStateBackend中(FsStateBackend不必用户特别指明,只需在初始化时传入HDFS 或本地路径即可,如new RocksDBStateBackend("hdfs:///hacluster/checkpoint")或new RocksDBStateBackend("file:///Data"))。

如果用户使用自定义窗口(window),不推荐用户使用RocksDBStateBackend。在自定义窗口中,状态以ListState的形式保存在StatBackend中,如果一个key值中有多个value值,则RocksDB读取该种ListState非常缓慢,影响性能。用户可以根据应用的具体情况选择FsStateBackend+HDFS或RocksStateBackend+HDFS。

4) 语法val env = StreamExecutionEnvironment．getExecutionEnvironment()
// start a checkpoint every 1000 ms
env．enableCheckpointing(1000)
// advanced options:
// 设置checkpoint的执行模式,最多执行一次或者至少执行一次
env．getCheckpointConfig．setCheckpointingMode(CheckpointingMode．EXACTLY_ONCE)
// 设置checkpoint的超时时间
env．getCheckpointConfig．setCheckpointTimeout(60000)
// 如果在只做快照过程中出现错误,是否让整体任务失败:true是  false不是
env．getCheckpointConfig．setFailTasksOnCheckpointingErrors(false)
//设置同一时间有多少 个checkpoint可以同时执行
env．getCheckpointConfig．setMaxConcurrentCheckpoints(1)
5) 修改State Backend的两种方式

第一种:单任务调整

修改当前任务代码

env．setStateBackend(new FsStateBackend("hdfs://namenode:9000/flink/checkpoints"));

或者new MemoryStateBackend()

或者new RocksDBStateBackend(filebackend, true);【需要添加第三方依赖】

第二种:全局调整

修改flink-conf．yaml

state．backend: filesystem

state．checkpoints．dir: hdfs://namenode:9000/flink/checkpoints

注意:state．backend的值可以是下面几种:jobmanager(MemoryStateBackend), filesystem(FsStateBackend), rocksdb(RocksDBStateBackend)

6) Checkpoint的高级选项

默认checkpoint功能是disabled的,想要使用的时候需要先启用checkpoint开启之后,默认的checkPointMode是Exactly-once

//配置一秒钟开启一个checkpoint
env．enableCheckpointing(1000)
//指定checkpoint的执行模式
//两种可选:
//CheckpointingMode．EXACTLY_ONCE:默认值
//CheckpointingMode．AT_LEAST_ONCE
env．getCheckpointConfig．setCheckpointingMode(CheckpointingMode．EXACTLY_ONCE)
一般情况下选择CheckpointingMode．EXACTLY_ONCE,除非场景要求极低的延迟(几毫秒)
注意:如果需要保证EXACTLY_ONCE,source和sink要求必须同时保证EXACTLY_ONCE
//如果程序被cancle,保留以前做的checkpoint
env．getCheckpointConfig．enableExternalizedCheckpoints(ExternalizedCheckpointCleanup．RETAIN_ON_CANCELLATION)
默认情况下,检查点不被保留,仅用于在故障中恢复作业,可以启用外部持久化检查点,同时指定保留策略:
ExternalizedCheckpointCleanup．RETAIN_ON_CANCELLATION:在作业取消时保留检查点,注意,在这种情况下,您必须在取消后手动清理检查点状态
ExternalizedCheckpointCleanup．DELETE_ON_CANCELLATION:当作业在被cancel时,删除检查点,检查点仅在作业失败时可用
//设置checkpoint超时时间
env．getCheckpointConfig．setCheckpointTimeout(60000)
//Checkpointing的超时时间,超时时间内没有完成则被终止
//Checkpointing最小时间间隔,用于指定上一个checkpoint完成之后
//最小等多久可以触发另一个checkpoint,当指定这个参数时,maxConcurrentCheckpoints的值为1
env．getCheckpointConfig．setMinPauseBetweenCheckpoints(500)
//设置同一个时间是否可以有多个checkpoint执行
env．getCheckpointConfig．setMaxConcurrentCheckpoints(1)
指定运行中的checkpoint最多可以有多少个
env．getCheckpointConfig．setFailOnCheckpointingErrors(true)
用于指定在checkpoint发生异常的时候,是否应该fail该task,默认是true,如果设置为false,则task会拒绝checkpoint然后继续运行
2． Flink的重启策略

Flink支持不同的重启策略,这些重启策略控制着job失败后如何重启。集群可以通过默认的重启策略来重启,这个默认的重启策略通常在未指定重启策略的情况下使用,而如果Job提交的时候指定了重启策略,这个重启策略就会覆盖掉集群的默认重启策略。

1) 概览

默认的重启策略是通过Flink的 flink-conf．yaml 来指定的,这个配置参数 restart-strategy 定义了哪种策略会被采用。如果checkpoint未启动,就会采用 no restart 策略,如果启动了checkpoint机制,但是未指定重启策略的话,就会采用 fixed-delay 策略,重试 Integer．MAX_VALUE 次。请参考下面的可用重启策略来了解哪些值是支持的。

每个重启策略都有自己的参数来控制它的行为,这些值也可以在配置文件中设置,每个重启策略的描述都包含着各自的配置值信息。

重启策略重启策略值Fixed delayfixed-delayFailure ratefailure-rateNo restartNone

除了定义一个默认的重启策略之外,你还可以为每一个Job指定它自己的重启策略,这个重启策略可以在 ExecutionEnvironment 中调用 setRestartStrategy() 方法来程序化地调用,注意这种方式同样适用于 StreamExecutionEnvironment。

下面的例子展示了如何为Job设置一个固定延迟重启策略,一旦有失败,系统就会尝试每10秒重启一次,重启3次。

val env = ExecutionEnvironment．getExecutionEnvironment()
env．setRestartStrategy(RestartStrategies．fixedDelayRestart(
 3, // 重启次数
 Time．of(10, TimeUnit．SECONDS) // 延迟时间间隔
))
2) 固定延迟重启策略(Fixed Delay Restart Strategy)

固定延迟重启策略会尝试一个给定的次数来重启Job,如果超过了最大的重启次数,Job最终将失败。在连续的两次重启尝试之间,重启策略会等待一个固定的时间。

重启策略可以配置flink-conf．yaml的下面配置参数来启用,作为默认的重启策略:

restart-strategy: fixed-delay
配置参数描述默认值restart-strategy．fixed-delay．attempts在Job最终宣告失败之前,Flink尝试执行的次数1,如果启用checkpoint的话是Integer．MAX_VALUErestart-strategy．fixed-delay．delay延迟重启意味着一个执行失败之后,并不会立即重启,而是要等待一段时间。akka．ask．timeout,如果启用checkpoint的话是1s

例子:

restart-strategy．fixed-delay．attempts: 3
restart-strategy．fixed-delay．delay: 10 s

固定延迟重启也可以在程序中设置:

val env = ExecutionEnvironment．getExecutionEnvironment()
env．setRestartStrategy(RestartStrategies．fixedDelayRestart(
 3, // 重启次数
 Time．of(10, TimeUnit．SECONDS) // 重启时间间隔
))
3) 失败率重启策略

失败率重启策略在Job失败后会重启,但是超过失败率后,Job会最终被认定失败。在两个连续的重启尝试之间,重启策略会等待一个固定的时间。