大數據框架及流批一體怎么選？

作者：天涯咫尺TGH 2024-06-25 13:08:31

大數據

在日常生活中，我們通常會先把數據存儲在一張表中，然后再進行加工、分析，這里就涉及到一個時效性的問題。

背景

在日常生活中，我們通常會先把數據存儲在一張表中，然后再進行加工、分析，這里就涉及到一個時效性的問題。

場景一：如果我們處理以年、月為單位的級別的數據，針對這些大量數據的實時性要求并不高。

場景二：如果我們處理的是以天、小時，甚至分鐘為單位的數據，那么對數據的時效性要求就比較高。

在第二種場景下，如果我們仍舊采用傳統的數據處理方式，統一收集數據，存儲到數據庫中，之后在進行分析，就可能無法滿足時效性的要求。

數據的計算模式主要分為：

批量計算(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();
    }
}