作者: 張皓
G7業務快覽
G7主要通過在貨車上的傳感器感覺車輛的軌迹、油耗、點熄火、載重、溫度等資料，将車輛、司機、車隊、貨主連接配接到一起，優化貨物運輸的時效、安全、成本等痛點問題。
整個資料是通過車載的傳感器裝置采集，比如公司的Smart盒子，CTBox盒子，油感裝置，溫度探頭等，将車輛資料上報到後端平台，在後端平台計算和處理，最後展示到使用者面前。
G7的業務場景是典型的IoT場景：
- 傳感器資料
- 資料種類多
- 資料品質差
- 資料低延遲
- 資料量大
其中，資料品質差的原因是整個鍊條會非常的長，從傳感器采集的車輛的資料，通過網絡營運商将資料上報到後端伺服器，再經過解析，mq，過濾，調用三方接口，業務處理，入庫，整個過程非常的長，造成資料在傳輸過程中出現資料重複，資料缺失等。另外一點，IoT場景需要資料傳輸的延遲非常低，比如進出區域報警，當車輛進入到某個電子圍欄中的時候需要觸發報警，這個時候需要快速産生報警事件，通常不能超過30s，否則時間太長車輛已經通過了某個電子圍欄區域再報警就沒有價值了。再一個，資料量也是非常大的，現在每天産生軌迹點20億+，每天産生資料量100億+，對計算性能的要求非常高。
實時計算選型
從上面的場景我們可以感覺到，在G7的IoT場景需要的是一個低延遲，處理速度快的實時計算引擎。最開始我們的一些架構是基于Lambda架構的，比如軌迹點計算，會使用實時計算引擎計算出實時資料，這份資料延遲比較低，但是資料不是很準确，另外需要用離線批量再計算一遍，這份資料通常比較準确，可以用來修複實時資料。這樣做的缺點也比較明顯，一是程式需要維護兩套代碼：實時程式和離線程式，二是實時資料不準确，準确的資料延遲又太高。後來我們驚喜的發現一種基于實時處理的架構體系Kappa。
Kappa的架構是強調資料的實時性，為了保證資料的實時性有些延遲太多的資料它會建議丢棄，所有的計算邏輯隻有在實時計算中，整個計算隻有一套邏輯，資料從MQ中擷取，經過資料處理層計算和加工，最後落入到資料存儲層，對外提供資料查詢功能。相對Lambda架構，Kappa架構更加适合IoT領域。
針對Kappa架構，我們對行業主流的實時流計算架構進行了對比：
分别對主流的流計算架構：Storm，Storm Trident，Spark Streaming，Google Cloud Dataflow，Flink做了對比。基于微批量的Spark Streaming和Storm Trident延遲比較高，從這點就不适合我們的場景。Storm的延遲很低，但是資料一緻性是At Least once，容錯機制比較複雜，流控會比較抖動，這些方面都是不太适合。其中，Flink的一緻性保證（1.4版本還支援了end-to-end一緻性），延遲比較低，容錯機制的開銷是比較小的（基于Chandy-Lamport的分布式快照），流控是比較優雅的（算子之間的資料傳輸是基于分布式記憶體阻塞隊列），應用邏輯與容錯是分離的（算子處理和分布式快照checkpoint），基于以上我們認為flink是比較适合IoT這個場景的。
G7業務應用案例
Flink目前在G7的應用場景，主要有三方面：
- 實時計算
- 實時ETL
- 統計分析
下面分别介紹下以上三個場景的使用。
在G7的場景中，有很多業務都屬于實時計算的範疇，比如進出區域事件，超速事件，怠速事件，超速事件，疲勞報警事件，危險駕駛報警，油耗計算，裡程計算等。其中疲勞報警計算是最早開始嘗試使用flink來落地的。
疲勞報警業務模型
這是G7針對客戶推出的G7大屏，其中風險相關的部分是根據疲勞計算得出。
根據G7的大資料計算，因為疲勞駕駛造成貨車事故的比重占到整個事故的20%。對疲勞駕駛進行報警和預警就顯得特别重要，可以有效降低事故發生的可能性。
根據車輛行駛的裡程，駕駛員行駛的裡程，駕駛時長，判斷是否存在疲勞駕駛。如果超過報警閥值則報警，如果在報警閥值下面在預警閥值上面則預警。報警和預警都是下發語音到貨車駕駛室提醒司機。
這個業務場景中面臨的最大挑戰是實時性，穩定性。隻有用最短的時間、最穩定的方式将告警下發到相關人員才能最大程度減少風險。
業務流程
在整個處理流程中，首先會去擷取疲勞配置，根據車輛的狀态資訊和司機打卡資訊與疲勞配置結合，判斷是否出現預警和報警。計算過程中會把疲勞駕駛開始的狀态緩存起來，疲勞駕駛結束的時候擷取之前的狀态資料，比對成功之後會生成一條完整的疲勞事件。中間會調用一些接口服務比如dubbo擷取車輛的配置資料、狀态資料，産生的疲勞報警則會調用下發語音的接口，疲勞事件結果也會存儲到hbase、mysql、kafka等。
Streaming模型
最後開發成Flink的程式，從頭到到尾分别由以下算子構成：消費kafka算子、類型轉換算子、資料過濾算子、異步調用第三方接口算子，視窗排序算子，疲勞處理業務邏輯算子，資料入庫算子組成。
這個過程，也是踩了不少坑，我們也有一些心得體會：
- 算子表達盡量單一
- 每個算子盡量内聚，算子間盡量低耦合
- 算子打散，異步+多線程的性能發揮更好
- 單獨設定每個算子單元的并行度，性能更優
- hash和balance根據情況選擇：隻有需要使用keyby和valuestate地方才使用hash重新分布資料。其他地方盡量使用balance并且上下遊并行度一緻，會将task串聯成一個線程，不會走網絡IO性能更高
- 使用Asynchronous I/O 調用dubbo接口，zuul，db，hbase等外部接口
有部分場景是資料簡單采集、處理，入庫，也就是實時ETL，包括從Kafka采集資料到HDFS、DB、HBase、ES、Kafka等，這部分工作可以抽象成Flink的算子表達：Source -> Transformation -> Sink。
這部分通常可以FlinkKafkaConumser、MapFunction、JDBCAppendTableSink這類代碼。如下：
有部分場景需要有一些實時的統計分析，比如統計最近一小時内全國各城市，車輛總數，司機總數，疲勞事件，進出區域事件，打卡次數，點熄火事件等。這種場景，通常可以使用Flink SQl的做實時分析，sql+視窗函數（固定視窗，滑動視窗）。代碼大緻如下：
實時計算平台開發和現狀
在業務上的成功落地，我們也希望能把打造一個實時計算平台，服務各條業務線，經過差不多3個月的打磨，内部代号為Glink的實時計算平台上線，大緻的架構如下：
Glink主要由以下部分組成：
- HDFS分布式檔案系統。用來存儲flink任務中産生的checkpoint/savepoint資料，任務報、第三方依賴包的存儲和分發，任務運作中産生的臨時資料等；
- Yarn統一計算資源平台。用來提供統一的分布式計算資源平台，任務送出，任務排程，任務執行，資源隔離功能。目前所有的flink任務都是通過yarn進行統一的計算資源管理；
- 性能監控AMP工具。使用點評開源的Cat，在此基礎上做二次開發并取名“天樞系統”。可以提供程式的耗時95、99線、平均耗時、最大耗時、java GC監控、線程監控、堆棧資訊等；
- 叢集監控管理。機器資源監控使用zabbix，提供cpu、記憶體、磁盤io、網絡io、連接配接數、句柄監控。叢集資源監控和管理使用開源Ambari，提供自動化安裝、配置、叢集整體任務、記憶體、cpu資源、hdfs空間、yarn資源大小監控報警；
- 任務監控報警。使用flink提供的statsD reporter将資料上傳導時序資料庫InfluxDB，通過掃描Infludb資料繪制出task的處理流量，通過監控流量閥值低于預期值報警；
- 診斷調試。使用成熟的日志查詢系統 es+logstash+kibana，通過采集每個節點的日志寫入到es中， 可以在kibana中查詢關鍵資訊擷取日志記憶體，提供診斷和調優程式的線索；
- Flink APP 程式應用層。具體開發的flink應用程式，通常解決實時etl，統計分析，業務計算的場景；
- Glink任務管控平台。将以下的功能進行封裝，提供統一的任務管理，運維管理功能。
實時計算平台展示-任務管理
實時計算平台展示-日志和性能監控
平台的部分功能介紹：
- 任務管理功能。提供任務釋出，修改，更新，停止，申請資源，資源稽核，啟動日志檢視功能；
- 運維管理功能。提供日志檢視，程式監控，任務監控，流量監控，異常報警等功能。
以上Glink實時計算平台的功能，基本上滿足使用者獨立完成從程式開發，釋出，調優，上線，運維的工作。
Glink-Framework開發架構
除了提供相應的平台功能，還需要在flink的生态上提供比較好的封裝和工具類，是以我們提供了開發工具的腳手架：Glink-Framework架構。
Glink-Framework提供以下封裝：
- 簡化pom檔案，減少大量的依賴、插件配置；
- 三方調用內建：dubbo，zuul；
- 三方資料庫內建：mysql，redis；
- 多環境管理；
- 依賴版本管理；
- 代碼監測工具：checkstyle，pmd，findbugs。
平台與業務方BP合作方式
另外一方面，我們認為flink是有一定的技術門檻，特别對于之前沒有并發程式設計、叢集開發經驗的小夥伴，需要有一段時間的學習才能上手，針對這個痛點，我們提出了技術BP的技術合作方式。我們會根據業務的複雜度，平台指派一至多名技術人員參與到業務方的整個開發和運維工作中，從需求分析到上線落地全程參與，後期還會有持續的技術分享和教育訓練幫助業務方學習開發能力。
踩坑
在整個平台化，以及業務開發的過程中，flink也踩坑不少，比較典型的下面一些。
- 并行度太多造成barrier對齊的花費時間更長，有個并行度28的子任務的對齊時間超過50s；
- Valuestate不能跨算子共享；
- flink1.3 kafka connector不支援partition增加；
- 與spring整合，出現handler比對的問題；
- hadoop的包沖突造成，程式無法正常啟動的問題且無異常；
其中比較有意思的是并行度太多，造成barrier對齊花費時間太多的問題。要了解這個問題首先要了解flink在生成checkpoint的過程中，會在source的插入barrier與正常消息一起往下遊發射，算子中等到指定的brrier後會觸發checkpoint。如下圖所：
這是在一個流的情況下，如果有多個流同時進入一個算子處理就會複雜一點。flink在做checkpoint的時候，發現有多個流進入一個算子，先進入這個算子的barrier對應的那段消息就會buffer到算子中等待另外的流對應的barrier也到達才會觸發checkpoint，這個buffer再等待的過程稱為checkpoint alignment（barrier對齊），如下圖：
線上上運作的某個程式的一些算子因為barrier對齊的時間超過50s，造成程式 checkpoint逾時失敗。對于這個問題，我們的調優政策是兩種，一是盡量減少并行度，就是讓流入一個算子的流盡量少，如果在4個以内barrier對齊的時間是比較少的。另外一種方式，使用at least once的語義替換exactly once的語義，這樣checkpoint的時候不會去做barrier對齊，資料到了算子馬上做checkpoint并發送下遊。目前 我們的解決辦法是根據不同的業務場景來區分，如果使用at least once資料保證就能滿足業務需求的盡量用at least once語義。如果不支援的，就減少并行度以此減少barrier對齊的資料量和時間。
平台收益
通過近段時間的平台化建設，在”降本增效“方面的收益主要展現在以下幾個方面：
- 資源使用率提高。目前通過對整個叢集的監控，在混合部署的情況下平均cpu使用率在20%左右，在某些cpu密集計算的業務cpu使用率會更高一些；
- 開發效率提升。比如ETL采集程式的開發，傳統開發采集資料、轉化、入庫大概需要1天左右時間，通過平台化的方式開發簡單的ETL程式在1小時内完成開發；
- 資料處理量大。平均每天處理資料量在80億條以上；
- 業務覆寫面廣。平台上線業務30+，預計年内突破100+。服務于公司各條業務線，IoT平台，EMS，FMS，智能挂車，企業解決方案，SaaS，硬體部門等。
未來規劃
未來對于flink的規劃，我們主要還是會圍繞“降本增效，提供統一的計算平台”為目标，主要聚焦在以下幾個方面：
1 .資源隔離更徹底。目前的資源隔離使用yarn的預設隔離方式隻是對記憶體隔離，後續需要使用yarn+cgroup對記憶體和cpu都做隔離。另外會考慮使用yarn的node label做徹底機器級别隔離，針對不同的業務劃分不同類型的機器資源，例如高CPU的任務對應CPU密集型的機器，高IO的任務對應IO比較好的機器；
- 平台易用性提高。平台包括代碼釋出、debug、調試、監控、問題排查，一站式解決問題；
- 減少Code。通過使用Flink SQL+UDF函數的方式，将常用的方法和函數進行封裝，盡量用SQL表達業務，提高開發效率。另外也會考慮CEP的模式比對支援，目前很多業務都可以用動态CEP去支援；
- 通用的腳手架。在Glink-Framework上持續開發，提供更多的source、sink、工具等，業務封裝，簡化開發；
此篇文章，摘自于張皓在 「Flink China社群線下 Meetup·成都站」 的技術分享