Kafka復制與Kafka Streams的實時數據處理案例分析

Apache Kafka 是一個分布式流處理平臺，用于構建實時數據管道和應用程序

1. Kafka 復制：

Kafka 復制是指將消息從一個主題（Topic）復制到另一個主題。這種復制可以用于多種場景，如數據備份、負載均衡或實現不同的數據處理需求。在 Kafka 中，復制是通過消費者（Consumer）和生產者（Producer）API 實現的。

案例：假設我們有一個名為 “input-topic” 的主題，我們希望將其中的數據復制到名為 “backup-topic” 的另一個主題。我們可以編寫一個簡單的 Kafka 消費者應用程序，從 “input-topic” 讀取數據，然后使用 Kafka 生產者將數據寫入 “backup-topic”。

2. Kafka Streams：

Kafka Streams 是一個用于處理實時數據流的庫，它允許你在 Kafka 集群上運行實時計算。Kafka Streams 提供了一個高級 API，可以方便地定義數據處理邏輯，如過濾、轉換、聚合等。

案例：假設我們有一個名為 “orders” 的主題，其中包含電子商務網站的訂單數據。我們希望實時計算每個產品類別的總銷售額。為此，我們可以使用 Kafka Streams 編寫一個實時數據處理應用程序。

以下是一個簡化的 Java 代碼示例：

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;

import java.util.Properties;

public class SalesAnalytics {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "sales-analytics");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> orders = builder.stream("orders");

        KTable<String, Double> salesByCategory = orders
                .mapValues(value -> parseOrder(value)) // 解析訂單數據
                .groupBy((key, order) -> order.getCategory()) // 按產品類別分組
                .reduce((order1, order2) -> order1.getAmount() + order2.getAmount(), Materialized.as("sales-by-category")); // 計算每個類別的總銷售額

        salesByCategory.toStream().to("sales-by-category-output", Produced.with(Serdes.String(), Serdes.Double()));

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}

在這個示例中，我們首先創建了一個 Kafka Streams 應用程序，然后從 “orders” 主題讀取訂單數據。接下來，我們對訂單數據進行解析、分組和聚合操作，最后將結果寫入名為 “sales-by-category-output” 的輸出主題。

總之，Kafka 復制和 Kafka Streams 都是實現實時數據處理的有效方法。Kafka 復制主要用于數據備份、負載均衡等場景，而 Kafka Streams 則提供了一個高級 API，用于實現更復雜的實時數據處理需求。