大數據框架及流批一體怎么選？
背景
在日常生活中，我們通常會先把數據存儲在一張表中，然后再進行加工、分析，這里就涉及到一個時效性的問題。
場景一：如果我們處理以年、月為單位的級別的數據，針對這些大量數據的實時性要求并不高。
場景二：如果我們處理的是以天、小時，甚至分鐘為單位的數據，那么對數據的時效性要求就比較高。
在第二種場景下，如果我們仍舊采用傳統的數據處理方式，統一收集數據，存儲到數據庫中，之后在進行分析，就可能無法滿足時效性的要求。
數據的計算模式主要分為：
- 批量計算(batch computing)、
- 流式計算(stream computing)、
- 交互計算(interactive computing)、
- 圖計算(graph computing)等。
其中，流式計算和批量計算是兩種主要的大數據計算模式，分別適用于不同的大數據應用場景。
流數據(或數據流)是指在時間分布和數量上無限的一系列動態數據集合體，數據的價值隨著時間的流逝而降低，因此必須實時計算給出秒級響應。流式計算，就是對數據流進行處理，是實時計算。
批量計算則統一收集數據，存儲到數據庫中，然后對數據進行批量處理的數據計算方式。兩者的區別主要體現在以下幾個方面：
(1)數據時效性不同
- 流式計算實時、低延遲；
- 批量計算非實時、高延遲。
(2)數據特征不同
- 流式計算的數據一般是動態的、沒有邊界的；
- 批處理的數據一般則是靜態數據。
(3)應用場景不同
- 流式計算應用在實時場景，時效性要求比較高的場景，如實時推薦、業務監控…。
- 批量計算一般說批處理，應用在實時性要求不高、離線計算的場景下，數據分析、離線報表等。
(4)運行方式不同
- 流式計算的任務持續進行的；
- 批量計算的任務則一次性完成。
流式計算框架平臺與相關產品
第一類，商業級流式計算平臺（IBM InfoSphere Streams、IBM StreamBase等）；
第二類，開源流式計算框架（Twitter Storm、S4等）；
第三類，公司為支持自身業務開發的流式計算框架。
(1)Strom：Twitter 開發的第一代流處理系統。
(2)Heron：Twitter 開發的第二代流處理系統。
(3)Spark streaming：是Spark核心API的一個擴展，可以實現高吞吐量的、具備容錯機制的實時流數據的處理。
(4)Flink：是一個針對流數據和批數據的分布式處理引擎。
(5)Apache Kafka：由Scala寫成。該項目的目標是為處理實時數據提供一個統一、高通量、低等待的平臺。
流式計算主要應用場景
流式處理可以用于兩種不同場景：事件流和持續計算。
(1)事件流
事件流具能夠持續產生大量的數據，這類數據最早出現與傳統的銀行和股票交易領域，也在互聯網監控、無線通信網等領域出現、需要以近實時的方式對更新數據流進行復雜分析如趨勢分析、預測、監控等。簡單來說，事件流采用的是查詢保持靜態，語句是固定的，數據不斷變化的方式。
(2)持續計算
比如對于大型網站的流式數據：網站的訪問PV/UV、用戶訪問了什么內容、搜索了什么內容等，實時的數據計算和分析可以動態實時地刷新用戶訪問數據，展示網站實時流量的變化情況，分析每天各小時的流量和用戶分布情況；比如金融行業，毫秒級延遲的需求至關重要。一些需要實時處理數據的場景也可以應用Storm，比如根據用戶行為產生的日志文件進行實時分析，對用戶進行商品的實時推薦等。
大數據流式計算可以廣泛應用于金融銀行、互聯網、物聯網等諸多領域，如股市實時分析、插入式廣告投放、交通流量實時預警等場景，主要是為了滿足該場景下的實時應用需求。數據往往以數據流的形式持續到達數據計算系統，計算功能的實現是通過有向任務圖的形式進行描述，數據流在有向任務圖中流過后，會實時產生相應的計算結果。整個數據流的處理過程往往是在毫秒級的時間內完成的。
通常情況下，大數據流式計算場景具有以下鮮明特征。
1)在流式計算環境中，數據是以元組為單位，以連續數據流的形態，持續地到達大數據流式計算平臺。數據并不是一次全部可用，不能夠一次得到全量數據，只能在不同的時間點，以增量的方式，逐步得到相應數據。
2)數據源往往是多個，在進行數據流重放的過程中，數據流中各個元組間的相對順序是不能控制的。也就是說，在數據流重放過程中，得到完全相同的數據流（相同的數據元組和相同的元組順序）是很困難的，甚至是不可能的。
3)數據流的流速是高速的，且隨著時間在不斷動態變化。這種變化主要體現在兩個方面，一個方面是數據流流速大小在不同時間點的變化，這就需要系統可以彈性、動態地適應數據流的變化，實現系統中資源、能耗的高效利用；另一方面是數據流中各個元組內容（語義）在不同時間點的變化，即概念漂移，這就需要處理數據流的有向任務圖可以及時識別、動態更新和有效適應這種語義層面上的變化。
4)實時分析和處理數據流是至關重要的，在數據流中，其生命周期的時效性往往很短，數據的時間價值也更加重要。所有數據流到來后，均需要實時處理，并實時產生相應結果，進行反饋，所有的數據元組也僅會被處理一次。雖然部分數據可能以批量的形式被存儲下來，但也只是為了滿足后續其他場景下的應用需求。
5)數據流是無窮無盡的，只要有數據源在不斷產生數據，數據流就會持續不斷地到來。這也就需要流式計算系統永遠在線運行，時刻準備接收和處理到來的數據流。在線運行是流式計算系統的一個常態，一旦系統上線后，所有對該系統的調整和優化也將在在線環境中開展和完成。
6)多個不同應用會通過各自的有向任務圖進行表示，并將被部署在一個大數據計算平臺中，這就需要整個計算平臺可以有效地為各個有向任務圖分配合理資源，并保證滿足用戶服務級目標。同時各個資源間需要公平地競爭資源、合理地共享資源，特別是要滿足不同時間點各應用間系統資源的公平使用。
什么是流批一體架構？
流處理和批處理都是常用的數據處理方式，它們各有優劣。流處理通常用于需要實時響應的場景，如在線監控和警報系統等。而批處理則通常用于離線數據分析和挖掘等大規模數據處理場景。選擇合適的處理方式取決于具體的業務需求和數據處理場景。
以前很多系統的架構都是采用的Lambda架構，它將所有的數據分成了三個層次：批處理層、服務層和速率層，每個層次都有自己的功能和目的。
- 批處理層：負責離線計算和歷史數據的存儲。
- 服務層：負責在線查詢和實時數據的處理。
- 速率層：負責對實時數據進行快速的處理和查詢。
這種架構，需要一套流處理平臺和一套批處理平臺，這就可能導致了一些問題：
- 資源浪費：一般來說，白天是流計算的高峰期，此時需要更多的計算資源，相對來說，批計算就沒有嚴格的限制，可以選擇凌晨或者白天任意時刻，但是，流計算和批計算的資源無法進行混合調度，無法對資源進行錯峰使用，這就會導致資源的浪費。
- 成本高：流計算和批計算使用的是不同的技術，意味著需要維護兩套代碼，不論是學習成本還是維護成本都會更高。
- 數據一致性：兩套平臺都是不一樣的，可能會導致數據不一致的問題。
因此，流批一體誕生了！
流批一體的技術理念最早是2015年提出的，初衷就是讓開發能用同一套代碼和API實現流計算和批計算，但是那時候實際落地的就少之又少，阿里巴巴在2020年雙十一首次實際落地。
Flink流批一體架構：
目前有哪些流處理的框架？
Kafka Stream
基于 Kafka 的一個輕量級流式計算框架，我們可以使用它從一個或多個輸入流中讀取數據，對數據進行轉換和處理，然后將結果寫入一個或多個輸出流中。
工作原理：讀取數據流 -> 數據轉換/時間窗口處理/狀態管理 -> 任務調度 -> 輸出結果
簡單示例：統計20秒內每個input的key輸入的次數，典型的例子：統計網站20秒內用戶的點擊次數。
public class WindowCountApplication {
private static final String STREAM_INPUT_TOPIC = "streams-window-input";
private static final String STREAM_OUTPUT_TOPIC = "streams-window-output";
public static void main(String[] args) {
Properties props = new Properties();
props.put(APPLICATION_ID_CONFIG, WindowCountApplication.class.getSimpleName());
props.put(BOOTSTRAP_SERVERS_CONFIG, KafkaConstant.BOOTSTRAP_SERVERS);
props.put(DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
StreamsBuilder builder = new StreamsBuilder();
builder.stream(STREAM_INPUT_TOPIC, Consumed.with(Serdes.String(), Serdes.String()))
.peek((key, value) -> Console.log("[input] key={}, value={}", key, value))
.groupByKey()
.windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofSeconds(20)))
.count()
.toStream()
.map((key, value) -> new KeyValue<>(key.key(), value))
.peek((key, value) -> Console.log("[output] key={}, value={}", key, value))
.to(STREAM_OUTPUT_TOPIC, Produced.with(Serdes.String(), Serdes.Long()));
KafkaStreams kStreams = new KafkaStreams(builder.build(), props);
Runtime.getRuntime().addShutdownHook(new Thread(kStreams::close));
kStreams.start();
}
}運行結果：{key}={value}，發送了3次A=1，2次B=1，以及1次C=1，統計結果在預期之內，即A出現3次，B出現2次，C出現1次。
Pulsar Function
和 Kafka Stream 類似，也是輕量級的流處理框架，不過它是基于 Pulsar 實現的一個流處理框架，同樣的，也是從一個或多個輸入流中讀取數據，對數據進行轉換和處理，然后將結果寫入一個或多個輸出流中。感興趣的可以參考我之前寫的文章：Pulsar Function簡介以及使用
工作原理：訂閱消息流 -> 處理消息 -> 發布處理結果
簡單示例：LocalRunner模式，按照逗號“,”去切分 input topic 的消息，然后轉換成數字進行求和，結果發送至 output topic。
public class IntSumFunction implements Function<String, Integer> {
public static final String BROKER_SERVICE_URL = "pulsar://localhost:6650";
public static final String INPUT_TOPIC = "persistent://public/default/int-sum-input";
public static final String OUTPUT_TOPIC = "persistent://public/default/int-sum-output";
public static final String LOG_TOPIC = "persistent://public/default/int-sum-log";
@Override
public Integer process(String input, Context context) {
Console.log("input: {}", input);
return Arrays.stream(input.split(","))
.map(Integer::parseInt)
.mapToInt(Integer::intValue)
.sum();
}
public static void main(String[] args) throws Exception {
FunctionConfig functionConfig = new FunctionConfig();
functionConfig.setName(IntSumFunction.class.getSimpleName());
functionConfig.setClassName(IntSumFunction.class.getName());
functionConfig.setRuntime(FunctionConfig.Runtime.JAVA);
functionConfig.setInputs(Collections.singleton(INPUT_TOPIC));
functionConfig.setOutput(OUTPUT_TOPIC);
functionConfig.setLogTopic(LOG_TOPIC);
LocalRunner localRunner = LocalRunner.builder()
.brokerServiceUrl(BROKER_SERVICE_URL)
.functionConfig(functionConfig)
.build();
localRunner.start(true);
}
}運行結果：1+2+3+4+5+6=21
Flink
- 一種流處理框架，具有低延遲、高吞吐量和高可靠性的特性。
- 支持流處理和批處理，并支持基于事件時間和處理時間的窗口操作、狀態管理、容錯機制等。
- 提供了豐富的算子庫和 API，支持復雜的數據流處理操作。
工作原理：接收數據流 -> 數據轉換 -> 數據處理 -> 狀態管理 -> 容錯處理 -> 輸出結果
簡單來說就是將數據流分成多個分區，在多個任務中并行處理，同時維護狀態信息，實現高吞吐量、低延遲的流處理。
簡單示例：從9966端口讀取數據，將輸入的句子用空格分割成多個單詞，每隔5秒做一次單詞統計。
public class WindowSocketWordCount {
private static final String REGEX = " ";
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStreamSource<String> socketTextStreamSource = env.socketTextStream("localhost", 9966);
SingleOutputStreamOperator<Tuple2<String, Integer>> streamOperator = socketTextStreamSource
.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (sentence, collector) -> {
for (String word : sentence.split(REGEX)) {
collector.collect(new Tuple2<>(word, 1));
}
})
.returns(Types.TUPLE(Types.STRING, Types.INT))
.keyBy(value -> value.f0)
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.sum(1);
streamOperator.print();
env.execute();
}
}運行結果：
Storm
- 一個開源的流處理引擎，旨在實現快速、可靠的數據流處理。
- 是業界最早出現的一個流處理框架（2011年），但是現在已經有許多其它優秀的流處理框架了，所以它在現在并不是唯一選擇。
工作原理：將數據流分成多個小的流（也稱為tuple），并將這些小流通過一系列的操作（也稱為bolt）進行處理。
簡單示例：在本地模式，使用Storm內置的RandomSentenceSpout充當數據源進行測試，用空格拆分生成的句子為多個單詞，統計每個單詞出現次數。
public class WindowedWordCountApplication {
public static void main(String[] args) throws Exception {
StreamBuilder builder = new StreamBuilder();
builder.newStream(new RandomSentenceSpout(), new ValueMapper<String>(0), 2)
.window(TumblingWindows.of(Duration.seconds(2)))
.flatMap(sentence -> Arrays.asList(sentence.split(" ")))
.peek(sentence -> Console.log("Random sentence: {}", sentence))
.mapToPair(word -> Pair.of(word, 1))
.countByKey()
.peek(pair -> Console.log("Count word: ", pair.toString()));
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("windowedWordCount", new Config(), builder.build());
Utils.sleep(20000);
cluster.shutdown();
}
}內置的RandomSentenceSpout隨機生成數據關鍵源代碼：
@Override
public void nextTuple() {
Utils.sleep(100);
String[] sentences = new String[]{
sentence("the cow jumped over the moon"), sentence("an apple a day keeps the doctor away"),
sentence("four score and seven years ago"), sentence("snow white and the seven dwarfs"), sentence("i am at two with nature")
};
final String sentence = sentences[rand.nextInt(sentences.length)];
LOG.debug("Emitting tuple: {}", sentence);
collector.emit(new Values(sentence));
}運行結果：隨機找一個單詞“nature”，統計的次數為10次。
Spark Streaming
基于 Spark API 的擴展，支持對實時數據流進行可擴展、高吞吐量、容錯的流處理。
工作原理：接收實時輸入數據流并將數據分成批次，然后由 Spark 引擎處理以批次生成最終結果流。
簡單示例：從 kafka 的 spark-streaming topic 讀取數據，按照空格“ ”拆分，統計每一個單詞出現的次數并打印。
public class JavaDirectKafkaWordCount {
private static final String KAFKA_BROKERS = "localhost:9092";
private static final String KAFKA_GROUP_ID = "spark-consumer-group";
private static final String KAFKA_TOPICS = "spark-streaming";
private static final Pattern SPACE = Pattern.compile(" ");
public static void main(String[] args) throws Exception {
Configurator.setRootLevel(Level.WARN);
SparkConf sparkConf = new SparkConf().setMaster("local[1]").setAppName("spark-streaming-word-count");
JavaStreamingContext streamingContext = new JavaStreamingContext(sparkConf, Durations.seconds(2));
Set<String> topicsSet = new HashSet<>(Arrays.asList(KAFKA_TOPICS.split(",")));
Map<String, Object> kafkaParams = new HashMap<>();
kafkaParams.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_BROKERS);
kafkaParams.put(ConsumerConfig.GROUP_ID_CONFIG, KAFKA_GROUP_ID);
kafkaParams.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
kafkaParams.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
JavaInputDStream<ConsumerRecord<String, String>> messages = KafkaUtils.createDirectStream(
streamingContext,
LocationStrategies.PreferConsistent(),
ConsumerStrategies.Subscribe(topicsSet, kafkaParams));
JavaDStream<String> linesStream = messages.map(ConsumerRecord::value);
JavaPairDStream<String, Integer> wordCountStream = linesStream
.flatMap(line -> Arrays.asList(SPACE.split(line)).iterator())
.mapToPair(word -> new Tuple2<>(word, 1))
.reduceByKey(Integer::sum);
wordCountStream.print();
streamingContext.start();
streamingContext.awaitTermination();
}
}運行結果：
如何選擇流處理框架？
- 簡單數據流處理
如果只是輕量級使用的話，可以結合技術棧使用消息中間件自帶的流處理框架就更節省成本。
- 使用的 Kafka 就用 Kafka Stream。
- 使用的 Pulsar 就用 Pulsar Function。
- 復雜數據流場景
綜上，可以結合數據規模、技術棧、處理延遲功能特性、未來的考慮、社區活躍度、成本和可用性等等進行選擇。
