假設一個 kafka 系統有以下兩個主題：
- 創建
- 已洗掉
它們用于宣傳專案的創建和洗掉。
kafka 中事件的結構是 JSON 并且對于兩個主題都是相同的：
{
"id" : "a1cf621a-2a96-4b70-9dd6-3c54a2819eef"
"timestamp": "2022-01-05T07:31:04.913000"
}
現在如何使用 spark (scala) 來累積已洗掉和創建的數量，以便我們通過時間戳獲得許多當前專案。
假設在kafka中發生以下事件
主題：創建
{"id":"1","timestamp":"2022-01-01T00:00:00.000000"}
{"id":"2","timestamp":"2022-01-02T00:00:00.000000"}
主題：已洗掉
{"id":"2","timestamp":"2022-01-03T00:00:00.000000"}
{"id":"1","timestamp":"2022-01-04T00:00:00.000000"}
所以這基本上意味著：
- 2022-01-01：創建了 1 個專案，專案總數為 1
- 2022-01-02：創建了 1 個專案，專案總數為 2
- 2022-01-03：1 個專案被洗掉，專案總數為 1
- 2022-01-04：1 個專案被洗掉，專案總數為 0
程式的結果輸出應該是每個時間戳的專案數，例如這里：
----------------------------------------
| timestamp | count |
----------------------------------------
| 2022-01-01T00:00:00.000000 | 1 |
| 2022-01-02T00:00:00.000000 | 2 |
| 2022-01-03T00:00:00.000000 | 1 |
| 2022-01-04T00:00:00.000000 | 1 |
----------------------------------------
如何合并兩個主題并按時間戳排序結果？
uj5u.com熱心網友回復：
您可以閱讀這兩個主題.option("subscribe", "created,deleted")
然后你有兩個主題的資料框，然后你可以決議時間戳的值，并且應該能夠按時間戳排序，并聚合/減少資料框以獲得輸出。
或者，Kafka 記錄中已經有一個時間戳，Spark 將其作為一列回傳。因此，您可以將生產者設計更改為具有一個主題，例如events，從值中洗掉時間戳，將 移至id記錄key，創建事件將具有非空值value，洗掉事件將具有null value.
不管怎樣，你仍然需要一個reducer 函式。
或者你不能使用Spark，并且按照建議的主題設計，那么在Kafka Streams/ksqlDB 中創建一個表就已經有了你想要的資料。不一定有時間戳資訊，但至少按 ID 或其他值匯總計數。
轉載請註明出處，本文鏈接：https://www.uj5u.com/yidong/496939.html