系統(tǒng)檢查點或作業(yè)快照
本文為您介紹實時計算Flink版系統(tǒng)檢查點或作業(yè)快照相關(guān)的常見問題。
兩次Checkpoint最小間隔時間計算方式
最小間隔時間是按照上次成功的Checkpoint開始計算。配置間隔時間為3，最小間隔時間為5，這種情況下，間隔時間會調(diào)整為5。
以兩個場景進(jìn)行說明，兩個場景Checkpoint間隔時間為3分鐘，超時時間為10分鐘，最小間隔時間為5分鐘。
場景一：作業(yè)正常運(yùn)行（Checkpoint每次都成功）
12:00第一次開始執(zhí)行Checkpoint，12:00:02 Checkpoint成功，第二次Checkpoint開始時間為12:05:02。
場景二：作業(yè)不正常（Checkpoint因某些原因超時或者失敗，本次場景以超時為例）
12:00第一次開始執(zhí)行Checkpoint，12:00:02 Checkpoint成功，12:05:02第二次開始執(zhí)行Checkpoint，12:15:02超時失敗，第三次Checkpoint開始時間為12:15:02。
Checkpoint最小間隔時間設(shè)置詳情請參見Tuning Checkpoint。
VVR 8.x和VVR 6.x使用的GeminiStateBackend有什么區(qū)別？
實時計算Flink版計算引擎VVR 6.x默認(rèn)使用V3版本的GeminiStateBackend，VVR 8.x默認(rèn)使用V4版本的GeminiStateBackend。
分類 | 詳情 |
基礎(chǔ)能力 |
|
狀態(tài)懶加載參數(shù) |
|
Managed Memory使用差異 | 僅在RSS（Resident Set Size）指標(biāo)上有區(qū)別：
說明 關(guān)于Managed Memory的更多解釋請參見TaskManager Memory。 |
關(guān)于GeminiStateBackend技術(shù)解讀詳情，請參見阿里云實時計算企業(yè)級狀態(tài)存儲引擎 Gemini 技術(shù)解讀。
從VVR 6.x升級到VVR 8.x時，遇到
You are using the new V4 state engine to restore old state data from a checkpoint報錯，解決方案請參見報錯：You are using the new V4 state engine to restore old state data from a checkpoint。
報錯：org.apache.flink.util.SerializedThrowable
報錯詳情
報錯原因
使用舊版Gemini在進(jìn)行快照時，有極小概率會遇到NullPointerException（NPE）報錯。該報錯通常是由于內(nèi)部內(nèi)存結(jié)構(gòu)引用為0，但尚未及時回收導(dǎo)致的。
解決方案
通常，該問題在系統(tǒng)運(yùn)行一段時間后或進(jìn)行重啟后，可以恢復(fù)正常。這個問題不會影響數(shù)據(jù)的正確性，只會導(dǎo)致Checkpoint失敗。您可以適當(dāng)增加Checkpoint失敗時的重啟容忍次數(shù)。
將VVR版本升級到8.0.1及以上版本，詳情請參見作業(yè)引擎版本升級。
報錯：You are using the new V4 state engine to restore old state data from a checkpoint
報錯詳情
從VVR 6.x升級到VVR 8.x時，報
You are using the new V4 state engine to restore old state data from a checkpoint。報錯原因
新版Gemini和舊版Gemini的Checkpoint無法兼容。
解決方案
您可以采用以下任何一種方式解決：
創(chuàng)建標(biāo)準(zhǔn)格式的作業(yè)快照并從該狀態(tài)啟動作業(yè)，詳情請參見手動創(chuàng)建作業(yè)快照和作業(yè)啟動。
無狀態(tài)重啟作業(yè)。
（不推薦）繼續(xù)使用舊版本Gemini。需要配置
state.backend.gemini.engine.type: STREAMING后重啟作業(yè)才能生效。參數(shù)配置方法請參見如何配置作業(yè)運(yùn)行參數(shù)？
報錯：No space left on device
報錯詳情
在作業(yè)運(yùn)行過程中，出現(xiàn)類似以下異常。
java.io.IOException: No space left on device at java.io.FileOutputStream.writeBytes(Native Method) ~[?:1.8.0_102] at java.io.FileOutputStream.write(FileOutputStream.java:326) ~[?:1.8.0_102] at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82) ~[?:1.8.0_102] at java.io.BufferedOutputStream.flush(BufferedOutputStream.java:140) ~[?:1.8.0_102] at java.io.FilterOutputStream.close(FilterOutputStream.java:158) ~[?:1.8.0_102] at org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.aliyun.oss.AliyunOSSOutputStream.close(AliyunOSSOutputStream.java:82) ~[?:?] at org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.FSDataOutputStream$PositionCache.close(FSDataOutputStream.java:72) ~[?:?] at org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.FSDataOutputStream.close(FSDataOutputStream.java:101) ~[?:?] at org.apache.flink.fs.osshadoop.common.HadoopDataOutputStream.close(HadoopDataOutputStream.java:52) ~[?:?] at com.alibaba.flink.statebackend.FlinkDataOutputStreamWapper.close(FlinkDataOutputStreamWapper.java:31) ~[flink-statebackend-gemini-2.1.23-vvr-3.0-SNAPSHOT.jar:2.1.23-vvr-3.0-SNAPSHOT] at com.alibaba.gemini.common.io.GeminiFileOutputViewImpl.close(GeminiFileOutputViewImpl.java:188) ~[flink-statebackend-gemini-2.1.23-vvr-3.0-SNAPSHOT.jar:2.1.23-vvr-3.0-SNAPSHOT] at com.alibaba.gemini.engine.filecache.InfiniteFileCache.lambda$flushBatchPages$1(InfiniteFileCache.java:635) ~[flink-statebackend-gemini-2.1.23-vvr-3.0-SNAPSHOT.jar:2.1.23-vvr-3.0-SNAPSHOT] at com.alibaba.gemini.engine.handler.GeminiEventExecutor.lambda$execute$1(GeminiEventExecutor.java:137) ~[flink-statebackend-gemini-2.1.23-vvr-3.0-SNAPSHOT.jar:2.1.23-vvr-3.0-SNAPSHOT] at com.alibaba.gemini.engine.handler.GeminiEventExecutor.doEventInQueue(GeminiEventExecutor.java:86) [flink-statebackend-gemini-2.1.23-vvr-3.0-SNAPSHOT.jar:2.1.23-vvr-3.0-SNAPSHOT] at com.alibaba.gemini.engine.handler.GeminiEventExecutor.run(GeminiEventExecutor.java:71) [flink-statebackend-gemini-2.1.23-vvr-3.0-SNAPSHOT.jar:2.1.23-vvr-3.0-SNAPSHOT] at org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989) [flink-dist_2.11-1.12-vvr-3.0.4-SNAPSHOT.jar:1.12-vvr-3.0.4-SNAPSHOT] at org.apache.flink.shaded.netty4.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) [flink-dist_2.11-1.12-vvr-3.0.4-SNAPSHOT.jar:1.12-vvr-3.0.4-SNAPSHOT] at java.lang.Thread.run(Thread.java:834) [?:1.8.0_102]報錯原因
本地磁盤空間不足。目前單Pod磁盤大小限制為20 GB，在該限制下，造成本地磁盤空間不足的原因通常有以下幾點：
狀態(tài)數(shù)據(jù)堆積。
計算節(jié)點的非狀態(tài)數(shù)據(jù)（例如日志）堆積。
異常導(dǎo)致的舊狀態(tài)數(shù)據(jù)堆積。
解決方案
您可以通過快照里的State Size判斷狀態(tài)數(shù)據(jù)是不是過大。如果確定因為狀態(tài)數(shù)據(jù)過大造成該報錯，則可以采用以下解決方案：
VVR 4.x和VVR 6.x版本優(yōu)化方案
您可以采用以下任何一種方案：
啟用存算分離功能（在VVR 4.0.12及以上版本已默認(rèn)啟用存算分離功能）
即配置state.backend.gemini.file.cache.type和state.backend.gemini.file.cache.preserved-space參數(shù)。詳情請參見存算分離配置。
增加并發(fā)度。
如果原來并發(fā)度是1，只能用一個pod，磁盤總空間是20 GB。如果并發(fā)加到4，作業(yè)就可以使用4個Pod，磁盤總空間相當(dāng)于是80 GB。
基于State TTL（Time To Live）進(jìn)行磁盤清理。
當(dāng)您設(shè)置了State TTL，如果時間超過了State的過期時間，則State數(shù)據(jù)就過期了，系統(tǒng)就會自動清理掉過期的State數(shù)據(jù)，即可釋放一定的磁盤空間。
VVR 3.x.x版本優(yōu)化方案
您可以采用以下任何一種方案：
對State進(jìn)行壓縮。
VVR 3.0.x版本配置state.backend.gemini.page.flush.local.compression: Lz4參數(shù)，本地State會有壓縮，能降低本地磁盤空間，但一定程度上會降低作業(yè)性能。
啟用存算分離功能。
VVR 3.0.3及以上版本配置state.backend.gemini.file.cache.type: LIMITED。本地盤會有一個18 GB的State限制，超過18 GB的數(shù)據(jù)會被存儲到遠(yuǎn)程DFS，下次讀取該部分?jǐn)?shù)據(jù)時，會從DFS讀取，相當(dāng)于把本地盤當(dāng)作一個本地文件緩存。
報錯：java.lang.IllegalArgumentException: Illegal Capacity: -1
報錯詳情
在作業(yè)使用Map State的遍歷操作時，在作業(yè)運(yùn)行過程中，有小概率會觸發(fā)到以下異常。
java.lang.IllegalArgumentException: Illegal Capacity: -1 at java.util.ArrayList.<init>(ArrayList.java:156) at com.alibaba.gemini.engine.pagestore.PageStoreImpl$3.<init>(PageStoreImpl.java:1113) at com.alibaba.gemini.engine.pagestore.PageStoreImpl.prefixIterator(PageStoreImpl.java:1094) at com.alibaba.gemini.engine.pagestore.PageStoreImpl.prefixIterator(PageStoreImpl.java:112) at com.alibaba.gemini.engine.table.BinaryKMapTable.internalEntries(BinaryKMapTable.java:83) at com.alibaba.gemini.engine.table.AbstractBinaryKMapTable.iterator(AbstractBinaryKMapTable.java:282) at com.alibaba.flink.statebackend.gemini.keyed.AbstractGeminiKeyedMapStateImpl.doIterator(AbstractGeminiKeyedMapStateImpl.java:496) at com.alibaba.flink.statebackend.gemini.keyed.AbstractGeminiKeyedMapStateImpl.iteratorWithMetrics(AbstractGeminiKeyedMapStateImpl.java:501) at com.alibaba.flink.statebackend.gemini.keyed.AbstractGeminiKeyedMapStateImpl.iterator(AbstractGeminiKeyedMapStateImpl.java:489) at com.alibaba.flink.statebackend.gemini.context.ContextMapState.entries(ContextMapState.java:97) at org.apache.flink.runtime.state.ttl.TtlMapState.entries(TtlMapState.java:107) at org.apache.flink.runtime.state.ttl.TtlMapState.entries(TtlMapState.java:102) at org.apache.flink.runtime.state.UserFacingMapState.entries(UserFacingMapState.java:77) at org.apache.flink.table.runtime.operators.join.stream.state.OuterJoinRecordStateViews$InputSideHasNoUniqueKey$1.<init>(OuterJoinRecordStateViews.java:279) at org.apache.flink.table.runtime.operators.join.stream.state.OuterJoinRecordStateViews$InputSideHasNoUniqueKey.getRecordsAndNumOfAssociations(OuterJoinRecordStateViews.java:276) at org.apache.flink.table.runtime.operators.join.stream.AbstractStreamingJoinOperator$AssociatedRecords.of(AbstractStreamingJoinOperator.java:229) at org.apache.flink.table.runtime.operators.join.stream.StreamingJoinOperator.processElement(StreamingJoinOperator.java:216) at org.apache.flink.table.runtime.operators.join.stream.StreamingJoinOperator.processRight(StreamingJoinOperator.java:134) at org.apache.flink.table.runtime.operators.join.stream.AbstractStreamingJoinOperator.processElement2(AbstractStreamingJoinOperator.java:136) at org.apache.flink.streaming.runtime.io.StreamTwoInputProcessorFactory.processRecord2(StreamTwoInputProcessorFactory.java:221) at org.apache.flink.streaming.runtime.io.StreamTwoInputProcessorFactory.lambda$create$1(StreamTwoInputProcessorFactory.java:190) at org.apache.flink.streaming.runtime.io.StreamTwoInputProcessorFactory$StreamTaskNetworkOutput.emitRecord(StreamTwoInputProcessorFactory.java:291) at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:134) at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:105) at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:66) at org.apache.flink.streaming.runtime.io.StreamTwoInputProcessor.processInput(StreamTwoInputProcessor.java:98) at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:423) at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:204) at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:684) at org.apache.flink.streaming.runtime.tasks.StreamTask.executeInvoke(StreamTask.java:639) at org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:650) at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:623) at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:779) at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566) at java.lang.Thread.run(Thread.java:834)報錯原因
產(chǎn)品已知缺陷，僅在VVR 4.0.10版本出現(xiàn)。
解決方案
將VVR版本升級到4.0.11及以上。
報錯：java.lang.NegativeArraySizeException
報錯詳情
當(dāng)作業(yè)使用了List State，在作業(yè)運(yùn)行過程中，會出現(xiàn)以下異常。
Caused by: java.lang.NegativeArraySizeException at com.alibaba.gemini.engine.rm.GUnPooledByteBuffer.newTempBuffer(GUnPooledByteBuffer.java:270) at com.alibaba.gemini.engine.page.bmap.BinaryValue.merge(BinaryValue.java:85) at com.alibaba.gemini.engine.page.bmap.BinaryValue.merge(BinaryValue.java:75) at com.alibaba.gemini.engine.pagestore.PageStoreImpl.internalGet(PageStoreImpl.java:428) at com.alibaba.gemini.engine.pagestore.PageStoreImpl.get(PageStoreImpl.java:271) at com.alibaba.gemini.engine.pagestore.PageStoreImpl.get(PageStoreImpl.java:112) at com.alibaba.gemini.engine.table.BinaryKListTable.get(BinaryKListTable.java:118) at com.alibaba.gemini.engine.table.BinaryKListTable.get(BinaryKListTable.java:57) at com.alibaba.flink.statebackend.gemini.subkeyed.GeminiSubKeyedListStateImpl.getOrDefault(GeminiSubKeyedListStateImpl.java:97) at com.alibaba.flink.statebackend.gemini.subkeyed.GeminiSubKeyedListStateImpl.get(GeminiSubKeyedListStateImpl.java:88) at com.alibaba.flink.statebackend.gemini.subkeyed.GeminiSubKeyedListStateImpl.get(GeminiSubKeyedListStateImpl.java:47) at com.alibaba.flink.statebackend.gemini.context.ContextSubKeyedListState.get(ContextSubKeyedListState.java:60) at com.alibaba.flink.statebackend.gemini.context.ContextSubKeyedListState.get(ContextSubKeyedListState.java:44) at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator.onProcessingTime(WindowOperator.java:533) at org.apache.flink.streaming.api.operators.InternalTimerServiceImpl.onProcessingTime(InternalTimerServiceImpl.java:289) at org.apache.flink.streaming.runtime.tasks.StreamTask.invokeProcessingTimeCallback(StreamTask.java:1435)報錯原因
List State中單個key對應(yīng)的State數(shù)據(jù)過大，即超過了2 GB。State數(shù)據(jù)過大產(chǎn)生的過程如下：
在作業(yè)正常運(yùn)行時，List State中單個Key下Append的Value會通過Merge進(jìn)行組合（例如在包含Window的List State中），State數(shù)據(jù)不斷累積。
在State數(shù)據(jù)累積到一定程度時，一開始會觸發(fā)OOM。而在作業(yè)從故障中恢復(fù)之后，List State的Merge過程會進(jìn)一步導(dǎo)致StateBackend申請的臨時Byte數(shù)組的大小超過可用的限制，從而出現(xiàn)該異常。
說明RocksDBStateBackend也會遇到類似的問題并觸發(fā)ArrayIndexOutOfBoundsException或者Segmentation fault。詳情請參見The EmbeddedRocksDBStateBackend。
解決方案
如果是Window算子導(dǎo)致的State數(shù)據(jù)過大，則建議減小窗口大小。
如果是作業(yè)邏輯不合理，則建議調(diào)整作業(yè)邏輯，例如將Key進(jìn)行拆分。
全量Checkpoint與增量Checkpoint的大小一致，是否正常？
如果您在使用Flink的情況下，觀察到全量Checkpoint與增量Checkpoint的大小一致，您需要：
檢查增量快照是否正常配置并生效。
是否為特定情況。在特定情況下，這種現(xiàn)象是正常的，例如：
在數(shù)據(jù)注入前（18:29之前），作業(yè)沒有處理任何數(shù)據(jù)，此時Checkpoint只包含了初始化的源（Source）狀態(tài)信息。由于沒有其他狀態(tài)數(shù)據(jù)，此時的Checkpoint實際上是一個全量Checkpoint。
在18:29時注入了100萬條數(shù)據(jù)。假設(shè)數(shù)據(jù)在接下來的Checkpoint間隔時間（3分鐘）內(nèi)被完全處理，并且期間沒有其他數(shù)據(jù)注入，此時發(fā)生的第一個增量Checkpoint將會包含這100萬條數(shù)據(jù)產(chǎn)生的所有狀態(tài)信息。
在這種情況下，全量Checkpoint和增量Checkpoint的大小一致是符合預(yù)期的。因為第一個增量Checkpoint需要包含全量數(shù)據(jù)狀態(tài)，以確保能夠從該點恢復(fù)整個狀態(tài)，導(dǎo)致它實際上也是一個全量Checkpoint。
增量Checkpoint通常是從第二個Checkpoint開始體現(xiàn)出來的，在數(shù)據(jù)穩(wěn)定輸入且沒有大規(guī)模的狀態(tài)變更時，后續(xù)的增量Checkpoint應(yīng)該顯示出大小上的差異，表明系統(tǒng)正常地只對狀態(tài)的增量部分進(jìn)行快照。如果仍然一致，則需要進(jìn)一步審查系統(tǒng)狀態(tài)和行為，確認(rèn)是否存在問題。