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