Flink開發中遇到的問題及解法

1. 當Source是Kafka的時候，如何設定Source Operator的并發度？

如果沒有指定，Source Operator的個數與叢集中的TaskManager的個數相等。如果手動設定，建議使用的slot個數=Kafka Partition的個數／TaskManager的個數。此時，Slot的個數需大于等于2.因為其中有一個Source Operator。也不建議在一個Slot中啟用多線程。

1. Barrier如果丢失了怎麼辦？

因為Barrier是從Source開始周期性的發送的，是以過一段時間未被标記為阻塞的input channel會收到下一個checkpoint的barrier，這時Flink會進行比對，發現如果目前的檢查點沒有完成，但下一個checkpoint已經過來了，那麼Flink會放棄目前的checkpoint，轉而使用下一個checkpoint。

1. 在Flink UI上Cancel Job，Job所有的任務都會停止嗎？

答：不是。Cancel按鈕隻是把Source，Transform和Sink這些Operator停掉，對應的線程停掉。但整個TaskManager還在。是以，如果Job中如果有不在Operator中初始化Spring容器，那麼即便Cancel Job以後，這些對象依然存在。是以，正确的姿勢是在Operator的open（）方法中初始化Spring容器。在close（）方法中釋放這些資源。

1. Job運作過程中TaskManager挂了怎麼辦？

如果TaskManager挂了，Flink會先将Job cancel掉。然後再以相同的JobID，往叢集中仍然存活的TaskManager上部署Job，這時候，如果還有足夠的task slot，則Job能夠恢複。但是這時候會有一個問題：部署在某些TaskManager上的Task數會比之前多，造成了這些TaskManager的負載較重，可能還是會出現問題。這時候就需要盡快恢複挂掉的TaskManager。

1. 某條資料在Input Channel之間傳輸失敗了怎麼辦？

會抛出Exception，然後Job會重新開機。

1. Flink讀取Kafka時，Checkpoint設定多久合适？

快照本身都是非常輕量級的，一般都在幾M或者幾十M。如果快照過大，比如幾百M甚至更多，就會對程式運作産生影響。官方給出的例子是幾秒鐘一次，具體可視Job情況決定。

1. Checkpoint和Savepoint有什麼差別？

savepoint可以了解為是一種特殊的checkpoint，savepoint就是指向checkpoint的一個指針，需要手動觸發，而且不會過期，不會被覆寫，除非手動删除。正常情況下的線上環境是不需要設定savepoint的。除非對job或叢集做出重大改動的時候，需要進行測試運作。

1. Flink的Operator不能帶成員變量？

Flink operator function中不能帶沒有實作flink序列化的成員變量。因為flink本身自己有一套序列化方式，在任務送出執行的時候會有validation，如果把沒有實作flink序列化的類作為成員變量，就會送出任務報錯。目前的解決方案是将operator function與實際業務邏輯分離。或将成員變為static。

1. 每個TaskManager設定多少個TaskSlot合适？

建議為CPU核數個。

1. TaskManager中的BufferPool不夠了咋辦？

需要增大配置項：taskmanager.network.numberOfBuffers的值，該值表示網絡棧buffer的數量，它的大小表示在同一時刻該TaskManager能夠擁有的流處理的資料交換的channel數。

1. Job運作中出現了OOM

說明保留的空間不夠，這時需減少中間層的空間大小，通過配置降低taskmanager.memory.fraction的值來減少中間層的記憶體占比。該值表示Flink用于管理底層buffer所占用的記憶體比例。

1. Job的并行度如何設定？

将所有的transform operator和sink operator的parallism設定成一樣的，source operator的parallism根據source而定。這樣的話，flink會自動把transform operator和sink operator 都merge成一個piple line去運作。那麼這時候一個job就變成隻有兩個operator了，source operator和merge後的operator，這個pipeline operator中間就沒有buffer了，性能最優。