狀態的基礎概念
1.State狀態
Flink實時計算程式為了保證計算程序中,出現例外可以容錯,就要將中間的計算結果資料存盤起來,這些中間資料就叫做State
State可以是多種型別的,默認是保存在JobManager的記憶體中,也可以保存到TaskManager本地檔案系統或HDFS這樣的分布式檔案系統
2.StateBackEnd
用來保存State的存盤后端就叫做StateBackEnd,默認是保存在JobManager的記憶體中,也可以保存在本地檔案系統或HDFS這樣的分布式檔案系統
3.CheckPointing
Flink實時計算為了容錯,可以將中間資料定期保存起來,這種定期觸發保存中間結果的機制叫CheckPointing. CheckPointing是周期執行的,具體的程序是JobManager定期的向TaskManager中的SubTask發送PRC訊息,SubTask將其計算的State保存到StateBackEnd中,并且向JobManager回應CheckPointing是否成功.如果程式出現例外或重啟,TaskManager中的SubTask可以從上一次成功的CheckPointing的State恢復
4.重啟策略
Flink實時計算程式,為了容錯,需要開啟CheckPointing,一旦開啟CheckPointing,如果沒有重啟策略,默認的重啟策略是無限重啟,也可以設定其他重啟策略.如:重啟固定次數卻可以延遲執行的策略
5.CheckPointingMode
exactly-once 精確一次性語意,可以保證資料消費且消費一次,但是要結合對應的資料源,比如Kafka支持exactly-one
at-least-once 至少消費一次,可能會重復消費,但是效率要比exactly-once高
ValueSate
/**
*
* ValueState的底層實作
* Flink中的State分為兩種:KeyedState(keyBy之后對應的State),和OperatorState(沒有keyBy的State)
*
* ValueState是KeyedState中的一種
*
* 1.KeyedState是KeyedState中的一種
* 2.如果想要容錯,必須要開啟checkpointing,并且按照Flink的狀態API進行編程(將中間結果保存在Flink特殊的變數中)
*
* 使用Flink的ValueState編程API實作可容錯的WordCount的功能
*/
public class ValueStateDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//開啟checkpointing checkpointing的默認重啟策略是無限重啟(Long型別的最大值)
env.enableCheckpointing(10000);
//開啟重啟策略 可以重啟3次 間隔5秒后開始重啟
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,Time.seconds(5)));
//呼叫Source讀取資料
DataStreamSource<String> lines = env.socketTextStream("linux01", 7777);
SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = lines.map(new MapFunction<String, Tuple2<String, Integer>>() {
@Override
public Tuple2<String, Integer> map(String line) throws Exception {
if (line.startsWith("error")){
throw new RuntimeException("輸入的資料錯誤,拋出例外");
}
String[] fields = line.split(" ");
return Tuple2.of(fields[0],1);
}
});
//按照單詞keyBy
KeyedStream<Tuple2<String, Integer>, String> keyedStream = wordAndOne.keyBy(tp -> tp.f0);
SingleOutputStreamOperator<Tuple2<String, Integer>> result = keyedStream.map(new RichMapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {
private ValueState<Integer> valueState;
//open方法中初始化或恢復狀態
@Override
public void open(Configuration parameters) throws Exception {
//構建狀態描述器
ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<Integer>("wc_value_state", Types.INT);
//初始化或者恢復狀態
valueState = getRuntimeContext().getState(stateDescriptor);
}
@Override
public Tuple2<String, Integer> map(Tuple2<String, Integer> input) throws Exception {
//獲取狀態中的歷史值
Integer history = valueState.value();
Integer current = input.f1;
//如果狀態中的歷史值為null 說明這個key第一次進入磁區
if (history == null){
history = 0;
}
current += history ;
//更新狀態
valueState.update(current);
input.f1=current;
return input;
}
});
result.print();
env.execute();
}
}MapState
/**
*
* 輸入如下資料, 將每個省每個城市中的錢進行累加
* 遼寧省,沈陽市,8000
* 遼寧省,大連市,7000
* 遼寧省,鞍山市,6000
* 遼寧省,鞍山市,8000
*
*/
public class MapStateDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//開啟checkpointing
env.enableCheckpointing(10000);
//設定重啟策略 錯誤率重啟策略（在一段時間內可也重啟指定的次數，如果超過時間范圍，重新計數）
env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.seconds(30),Time.seconds(5)));
//呼叫Source讀取資料
DataStreamSource<String> lines = env.socketTextStream("linux01", 7777);
//對資料進行整理
//遼寧省,沈陽市,8000
SingleOutputStreamOperator<Tuple3<String, String, Integer>> tpStream = lines.map(new MapFunction<String, Tuple3<String, String, Integer>>() {
@Override
public Tuple3<String, String, Integer> map(String line) throws Exception {
String[] fields = line.split(",");
String province = fields[0];
String city = fields[1];
int money = Integer.parseInt(fields[2]);
return Tuple3.of(province,city,money);
}
});
//按照省份進行keyBy
KeyedStream<Tuple3<String, String, Integer>, String> keyedStream = tpStream.keyBy(tp -> tp.f0);
SingleOutputStreamOperator<Tuple3<String, String, Integer>> result = keyedStream.map(new CityAmountFunction());
result.print();
env.execute();
}
private static class CityAmountFunction extends
RichMapFunction<Tuple3<String, String, Integer>,Tuple3<String, String, Integer>>{
private MapState<String, Integer> mapState;
@Override
public void open(Configuration parameters) throws Exception {
//獲取狀態描述器
MapStateDescriptor<String, Integer> stateDescriptor =
new MapStateDescriptor<>("cityAmount_state", Types.STRING, Types.INT);
//初始化或恢復狀態
mapState = getRuntimeContext().getMapState(stateDescriptor);
}
@Override
public Tuple3<String, String, Integer> map(Tuple3<String, String, Integer> input) throws Exception {
String city = input.f1;
Integer current = input.f2;
Integer history = mapState.get(city);
//如果歷史值為null 說明這個city第一次進入磁區
if (history == null){
history = 0;
}
current += history;
//更新狀態
mapState.put(city,current);
input.f2=current;
return input;
}
}
}ListState
/**
*
*輸入如下資料
* u001,view
* u001,pay
* u001,view
* u001,view
* u001,pay
* u002,view
* 統計每個用戶最近5件event
*/
public class ListStateDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//開啟checkpointing
env.enableCheckpointing(10000);
//呼叫Source讀取資料
DataStreamSource<String> lines = env.socketTextStream("linux01", 7777);
//整理資料
SingleOutputStreamOperator<Tuple2<String, String>> tpStream
= lines.map(line -> Tuple2.of(line.split(",")[0], line.split(",")[1]))
.returns(new TypeHint<Tuple2<String, String>>() {});
//按照用戶keyBy
KeyedStream<Tuple2<String, String>, String> keyedStream = tpStream.keyBy(tp -> tp.f0);
SingleOutputStreamOperator<Tuple2<String, List<String>>> result = keyedStream.map(new UserEventFunction());
result.print();
env.execute();
}
private static class UserEventFunction extends RichMapFunction<Tuple2<String, String>,Tuple2<String, List<String>>>{
private ListState<String> listState;
@Override
public void open(Configuration parameters) throws Exception {
//創建狀態描述器
ListStateDescriptor<String> stateDescriptor = new ListStateDescriptor<>("userEvent_state", Types.STRING);
//初始化或恢復狀態
listState = getRuntimeContext().getListState(stateDescriptor);
}
@Override
public Tuple2<String, List<String>> map(Tuple2<String, String> input) throws Exception {
String event = input.f1;
listState.add(event);
ArrayList<String> events = (ArrayList<String>)listState.get();
if (events.size()>5){
events.remove(0);
}
return Tuple2.of(input.f0,events);
}
}
}
簡單案例--定義兩個狀態
/**
*
* user01, activity01, view
* user01,activity01,join
* user01,activity02,view
* user02,activity02,view
* user02,activity02,view
* user03,activity02,view
* user02,activity02,join
* user03,activity01,view
*
*
* 實時統計出各個活動，各種事件的次數和人數(次數出現就累計，人數要按照用戶ID去重)
* activity01,view,2,2
* activity01,join,1,1
* activity02,view,4,3
* activity02,join,1,1
*/
public class ActivityCountDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//開啟checkpointing
env.enableCheckpointing(10000);
//讀取資料
DataStreamSource<String> lines = env.socketTextStream("linux01", 7777);
//對資料進行整理
//user01, activity01, view
SingleOutputStreamOperator<Tuple3<String, String, String>> tpStream
= lines.map(new MapFunction<String, Tuple3<String, String, String>>() {
@Override
public Tuple3<String, String, String> map(String line) throws Exception {
String[] fields = line.split(",");
String uid = fields[0];
String aid = fields[1];
String event = fields[2];
return Tuple3.of(uid, aid, event);
}
});
//按斬訓動ID和事件進行keyBy
KeyedStream<Tuple3<String, String, String>, Tuple2<String, String>> keyedStream
= tpStream.keyBy(new KeySelector<Tuple3<String, String, String>, Tuple2<String, String>>() {
@Override
public Tuple2<String, String> getKey(Tuple3<String, String, String> tp) throws Exception {
return Tuple2.of(tp.f1, tp.f2);
}
});
SingleOutputStreamOperator<Tuple4<String, String, Integer, Integer>> result = keyedStream.process(new ActivityCountFunction());
result.print();
env.execute();
}
private static class ActivityCountFunction
extends KeyedProcessFunction<Tuple2<String, String>,Tuple3<String, String, String>, Tuple4<String, String, Integer,Integer>> {
private ValueState<Integer> countState;
private ValueState<HashSet<String>> uidState;
@Override
public void open(Configuration parameters) throws Exception {
//獲取狀態描述器
ValueStateDescriptor<Integer> countStateDescriptor = new ValueStateDescriptor<Integer>("count_state", Types.INT);
ValueStateDescriptor<HashSet<String>> uidStateDescriptor = new ValueStateDescriptor<>("uid_state",
TypeInformation.of(new TypeHint<HashSet<String>>() {}));
//初始化或恢復狀態
countState = getRuntimeContext().getState(countStateDescriptor);
uidState = getRuntimeContext().getState(uidStateDescriptor);
}
@Override
public void processElement(Tuple3<String, String, String> input, Context ctx,
Collector<Tuple4<String, String, Integer, Integer>> out) throws Exception {
//統計次數
Integer historyCount = countState.value();
if (historyCount == null){
historyCount = 0;
}
historyCount += 1;
//更新狀態
countState.update(historyCount);
//統計人數
HashSet<String> historyUids = uidState.value();
if (historyUids == null){
historyUids =new HashSet<>();
}
historyUids.add(input.f0);
//更新狀態
uidState.update(historyUids); //參考型別可以不用更新
//輸出
out.collect(Tuple4.of(ctx.getCurrentKey().f0,ctx.getCurrentKey().f1,historyCount,historyUids.size()));
}
}
}轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/430377.html
標籤：其他