場景：
RDD<JSONObject>，JSONObject裡面有TBNAME欄位和PAYLOAD欄位，分別代表表名和原始日誌內容
需要1.在原始內容里加入系統時間欄位 2.按表名取系統時間逆序取前100條入庫
30s時間視窗,處理2w條資料；4張表，但打的資料均為1張表的資料
以下我說明的時間都是有資料的表的處理時間（1張表有資料處理，在過其他表時也需要filter表名，需要耗時；所以總體的批處理時間會比我描述的時間要大）

資源：executor-cores:4,core-memory:2g,driver-memory:512m（沒辦法，資源要省著用，目前這個資源對執行環境而言已經算奢侈了）

PS:機器為虛擬機器，所以這裡的核的計算能力與實體機的計算能力是不能相提並論的（是一臺實體機虛擬了5個虛擬機器組裝的叢集）

a.實現方式
for迴圈表名->按表名filter->並map(取PAYLOAD欄位)->takeOrder(num[取的條數],comparator[按時間逆序])
結果：27~30+s


b.實現方式
filter(按需要取的所有表名)->mapToPair(Tuple2<表名，PAYLOADJSONObject[已經加上系統時間欄位]>)->groupByKey(得到JavaRDD<表名,Iterator>)->mapToPair(從Iterator取出JSONObject按系統時間排序,sublist得到需要的數目的JSONObject)->collectMap
結果：會導致Java Heap OutOfMemory


c.實現方式
filter(按需要取的所有表名)->mapToPair(Tuple2<表名，PAYLOADJSONObject[已經加上系統時間欄位]>)->for迴圈表名->JavaPairRDD按key filter->map只取value返回，得到JavaRDD<JSONObject[PAYLOAD]>->takeOrder
結果：47~1min+


以上RDD均是JavaRDD<JSONObject>或者JavaPairRDD<String,JSONObject>


d.實現方式
filter(按需要取的所有表名)-> 並map（取PAYLOAD欄位；組成string:tablaname#systemtime#BASE64.encode(JSONSTRING)(PAYLOAD含有SYSTEM欄位)，成為JavaRDD<byte[]>）->for表名->filter(按#第一個欄位過濾)->takeOrder(按#第二個欄位排序)->得到前100的list->按#第三個欄位生成List<JSONObject>
場景：11w+資料量（應該是之前資料量的5、6倍），處理時間 幾秒~十幾秒


雖然有本質的提高，但目前仍不滿足要求；還應該繼續除錯


e.實現方式（採用d的實現方式的基礎上），同時再進行優化
我們job使用的cores和kafka的partition的數目相差比較大，cores=4,parttions=16
所以考慮在DStream transform時，對於rdd進行coalesce(cores,false)---即對rdd進行重新分割槽保證和cores數目相同，而且shuffle=false，因為是partition的合併，不需要shuffle

場景：11w+資料量（應該是之前資料量的5、6倍），處理時間 5秒~7秒；此時記錄批處理時間比較有意義，批處理時間為11~15左右（4張表）

另外，還嘗試了用多執行緒分別過濾4張表，takeOrder併入庫；但由於機器的cores計算能力並不強，所以反而在併發執行時每個JOB的執行時間變長，併發執行完的整體時間和目前的執行整體時間差不多。所以在cores不多，且機器運算能力不強的情況下，不考慮使用此方法。

真實場景下資料會分散多個表的資料，每個Job的壓力會相應減少，執行時間會比較目前的執行時間短。總體時間也許短或者與現在一致。

f.實現方式(在d+e的基礎上)
mapPartitions(得到的Iterator按表名過濾，並取每張表時間最新的100條放入list中)[理論上每張表的100條應該就等於表在每個分割槽最新100條再取100條]->for迴圈->filter(按#第一個欄位過濾)->takeorder(按第二個欄位排序)->得到前100的list->按#第三個欄位生成list<JSONObject>
目前實驗結果中f方式最好，第一張表的時間較長需要7~9秒，2,3,4表在persist的基礎上取都花費毫秒級的時間，所以總處理時間10秒左右

總結一下：
1.每個表的最新一百條，等於這個表在每個分割槽上取最新一百條之後的集合中再取一百條；這樣就可以降低後面filter和takeorder的資料量
2.JavaRDD<JSONObject>序列化/反序列化代價很大，能用JavaRDD<byte[]>儘量用
3.如果core比較少且core計算能管理不強，縮小分割槽數目減少任務數；因為一個core併發執行多個任務，執行緒之間的切換也會造成耗時

重大發現，之前persist RDD時的觸發action是isempty，isempty判斷會短路判斷（比如第一個分割槽有資料就不再繼續往下判斷），那麼RDD persist的只有1個分割槽，時間又長，而且不是全部分割槽能夠被後續複用；改為rdd.count()==0,判斷完成後，RDD的4個分割槽均可以被複用。後面的takeorder job在使用f方法的情況下4個加起來耗時1秒左右。非常振奮人心，不僅減少了實時監控的耗時，而且這個RDD還可以被後面的氣泡圖和黑白灰用。

截圖留念：