筆者的項目，一直是用flink進行流處理，為什么這次寫flink批處理離線計算呢，因為客戶提的新需求更適合用批處理離線計算:
客戶不需要實時計算的結(jié)果，認為實時結(jié)果對他們沒有參考意義
客戶要求計算一個月(或更長時間)內(nèi)歷史數(shù)據(jù)的基準值，認為一個月(或更長時間)歷史數(shù)據(jù)得到的基準值才能更加準確的評估和預測未來的趨勢
大數(shù)據(jù)計算分為離線計算和實時計算，離線計算就是我們通常說的批計算，代表是Hadoop MapReduce、Hive等大數(shù)據(jù)技術(shù)，實時計算也被稱作流計算，代表是Storm、Spark Streaming、Flink等大數(shù)據(jù)技術(shù)。
批處理在大數(shù)據(jù)世界有著悠久的歷史。批處理主要操作大容量靜態(tài)數(shù)據(jù)集，并在計算過程完成后返回結(jié)果。
批處理模式中使用的數(shù)據(jù)集通常符合下列特征：
有界：批處理數(shù)據(jù)集代表數(shù)據(jù)的有限集合
持久：數(shù)據(jù)通常始終存儲在某種類型的持久存儲位置中
大量：海量數(shù)據(jù)集
批處理非常適合需要訪問全套記錄才能完成的計算工作。例如在計算總數(shù)和平均數(shù)時，必須將數(shù)據(jù)集作為一個整體加以處理，而不能將其視作多條記錄的集合。這些操作要求在計算進行過程中數(shù)據(jù)維持自己的狀態(tài)。
需要處理大量數(shù)據(jù)的任務(wù)通常最適合用批處理操作進行處理。無論直接從持久存儲設(shè)備處理數(shù)據(jù)集，或首先將數(shù)據(jù)集載入內(nèi)存，批處理系統(tǒng)在設(shè)計過程中就充分考慮了數(shù)據(jù)的量，可提供充足的處理資源。由于批處理在應對大量持久數(shù)據(jù)方面的表現(xiàn)極為出色，因此經(jīng)常被用于對歷史數(shù)據(jù)進行分析。
相信各位已經(jīng)看過諸多流計算優(yōu)點的文章，流計算的優(yōu)點筆者就略過, 下面說一下批處理相比流處理的優(yōu)點。
由于批量和流式處理數(shù)據(jù)粒度不一樣，批量每次處理一定大小的數(shù)據(jù)塊（輸入一般采用文件系統(tǒng)），一個任務(wù)處理完一個數(shù)據(jù)塊之后，才將處理好的中間數(shù)據(jù)發(fā)送給下游。流式計算則是以單條記錄為單位，任務(wù)在處理完一條記錄之后，然后發(fā)送給下游進行處理。流式計算來一條記錄就計算一次，計算量巨大，當不需要中間值的時候，這種計算屬實浪費，因此批處理的吞吐量更大、資源利用率更高、系統(tǒng)的開銷更小。
精確一次(exactly once)是指數(shù)據(jù)處理沒有數(shù)據(jù)丟失和重復處理的現(xiàn)象。
流處理的數(shù)據(jù)來源一般是消息隊列，是無界的，數(shù)據(jù)是一條一條獲取，在加載數(shù)據(jù)時可能會出現(xiàn)網(wǎng)絡(luò)連接等問題，所以流處理需要解決數(shù)據(jù)丟失和重復處理的問題，實現(xiàn)精確一次(exactly once)的語義相對復雜，目前storm流框架目前不支持(exactly once)，spark為了支持(exactly once)引入預寫日志(AWL)并且offset由Spark自身管理 ，flink為了支持(exactly once)引入快照(snapshot)機制， 雖然流處理能夠解決數(shù)據(jù)丟失和重復計算問題，但需要引入各種機制，而這增加了系統(tǒng)消耗的資源。
批處理的數(shù)據(jù)源是靜態(tài)塊，比如文件，hdfs文件，批處理一次性加載一批數(shù)據(jù)，基本不會出現(xiàn)數(shù)據(jù)丟失和重復計算的情況。
流處理水印(watermark)忽略數(shù)據(jù)
如果說流處理引入各種機制增加資源消耗可以解決數(shù)據(jù)丟失和重復處理問題，那么對于亂序數(shù)據(jù)流則存在忽略數(shù)據(jù)的可能。
流處理數(shù)據(jù)沒有邊界，需要窗口(window)的概念，根據(jù)窗口來匯總計算。窗口(window)類型有很多種, 滾動窗口(Tumbling Window)、滑動窗口(Sliding Window)和會話窗口(Session window)等，窗口(window)中需要定義時間，流處理中存在事件時間(event time)和處理時間(process time)。對于亂序的數(shù)據(jù)，為此又引入了水印(watermark)機制。具體概念讀者自行查閱。
水印(watermark)有一個允許延時(allow lateness)的參數(shù), 窗口(window)接收到水印(watermark)后，再等待一段時間才會關(guān)閉窗口，如果這段時間有些數(shù)據(jù)依然沒有發(fā)送過來，那就只能忽略它們了。允許延時(allow lateness)參數(shù)設(shè)置的大，系統(tǒng)占用的資源就多，而且允許延時(allow lateness)的參數(shù)不能設(shè)置無限大，因此如果數(shù)據(jù)源異常亂序，流處理的窗口就等不到延時數(shù)據(jù)過來就進行匯總計算，導致延時數(shù)據(jù)未處理。
批處理數(shù)據(jù)有界，所有的數(shù)據(jù)全部都會加載，不用考慮數(shù)據(jù)源的順序，不會出現(xiàn)忽略數(shù)據(jù)的情況，也不需要窗口(window) ，時間，水印等機制。
通過上面簡單的介紹和對比，發(fā)現(xiàn)客戶的需求更適合用批處理離線計算，由于Flink是一個流處理框架， 可以處理有邊界和無邊界的數(shù)據(jù)流。無邊界的數(shù)據(jù)就是流數(shù)據(jù)，有邊界的數(shù)據(jù)就是批數(shù)據(jù)，因此Flink也是支持批處理的。所以筆者采用Flink進行批處理計算。
以下是核心代碼:
flink執(zhí)行環(huán)境：
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
從kafka中獲取DataSet數(shù)據(jù)源，DataSet表示批處理的數(shù)據(jù) (流處理是DataStream)：
DataSetString, String>> recordsDataSetDataSet = env.createInput(KafkaInputFormat
.buildKafkaInputFormat().setBootstrapServers(KafkaServers)
.setGroupId(xx).setTopic(sourceTopic).finish());
繼承GenericInputFormat類實現(xiàn)自定義獲取kafka數(shù)據(jù)源 KafkaInputFormat：
public class KafkaInputFormat extends GenericInputFormatString, String>> {
@Override
public void open(GenericInputSplit split) throws IOException {
consumer = new KafkaConsumer<String,String>(props);
initPartionMap();
}
//獲取kafka topic每個分區(qū)的偏移量,用做kafka消費結(jié)束的標識
void initPartionMap(){
Collection partitionInfos = consumer.partitionsFor(topic);
List tp =new ArrayList();
partitionInfos.forEach(partitionInfo -> {
tp.add(new TopicPartition(topic,partitionInfo.partition()));
consumer.assign(tp);
consumer.seekToEnd(tp);
partionOffsetMap.put(partitionInfo.partition(),consumer.position(new TopicPartition(topic, partitionInfo.partition())));
partionBooleanMap.put(partitionInfo.partition(), false);
//獲取參數(shù)值后返回最初
consumer.seekToBeginning(tp);
});
}
//消費kafka是否結(jié)束
@Override
public boolean reachedEnd() throws IOException {
return !partionBooleanMap.containsValue(false);
}
@Override
public ConsumerRecords<String, String> nextRecord(ConsumerRecords<String, String> reuse) {
//從kafka中獲取一批數(shù)據(jù)
final ConsumerRecords<String, String> records consumer.poll(Duration.ofMillis(pollTime));
for (ConsumerRecord<String, String> record : records) {
Integer partion=record.partition();
Long offset= record.offset();
//表示已有分區(qū)已經(jīng)消費完
if(offset+1==partionOffsetMap.get(partion)) {
partionBooleanMap.put(partion, true);
}
}
return records;
}
以前的流處理計算過程經(jīng)過批處理改造后，計算時間大大縮短，也不需要設(shè)置窗口(window)、等待時間(allow lateness)和水印(watermark)，而且計算完成程序自動退出，不再占用系統(tǒng)資源。
流處理和批處理都有各自的優(yōu)缺點和應用場景，應該根據(jù)項目需求選擇合適的。
更多精彩干貨分享
點擊下方名片關(guān)注
IT那活兒
