Twitter把Kafka當(dāng)作存儲系統(tǒng)使用,使用kafka的公司-ESG跨境

Twitter把Kafka當(dāng)作存儲系統(tǒng)使用,使用kafka的公司

Twitter把Kafka當(dāng)作存儲系統(tǒng)使用

當(dāng)開發(fā)者通過API消費Twitter的公共數(shù)據(jù)時，他們需要獲得可靠性、速度和穩(wěn)定性方面的保證。因此，在不久前，我們推出了Account Activity Replay API幫助開發(fā)者們提升他們系統(tǒng)的穩(wěn)定性。這個API是一個數(shù)據(jù)恢復(fù)工具，開發(fā)者可以用它來檢索最早發(fā)生在5天前的事件，恢復(fù)由于各種原因（包括在實時傳遞時突然發(fā)生的服務(wù)器中斷）沒有被傳遞的事件。

除了構(gòu)建API來提升開發(fā)者體驗，我們還做了一些優(yōu)化：

·提高Twitter內(nèi)部工程師的生產(chǎn)力。

·保持系統(tǒng)的可維護性。具體來說，就是盡量減少開發(fā)人員、站點可靠性工程師和其他與系統(tǒng)交互的人員的上下文切換。

基于這些原因，在構(gòu)建這個API所依賴的回放系統(tǒng)時，我們利用了Account Activity API現(xiàn)有的實時系統(tǒng)設(shè)計。這有助于我們重用現(xiàn)有的工作，并最小化上下文切換負擔(dān)和培訓(xùn)工作。

實時系統(tǒng)采用了發(fā)布和訂閱架構(gòu)。為了保持架構(gòu)的一致性，構(gòu)建一個可以讀取數(shù)據(jù)的存儲層，我們想到了傳統(tǒng)的流式技術(shù)——Kafka。

1

背景

兩個數(shù)據(jù)中心產(chǎn)生實時事件，事件被寫入到跨數(shù)據(jù)中心的主題上，實現(xiàn)數(shù)據(jù)冗余。

但并不是所有的事件都需要被傳遞，所以會有一個內(nèi)部應(yīng)用程序負責(zé)對事件進行篩選。這個應(yīng)用程序消費來自這些主題的事件，根據(jù)保存在鍵值存儲中的一組規(guī)則來檢查每一個事件，并決定是否應(yīng)該通過我們的公共API將消息傳遞給特定的開發(fā)者。事件是通過Webhook傳遞的，每個Webhook URL都有一個開發(fā)人員負責(zé)維護，并有唯一的ID標(biāo)識。

圖1：數(shù)據(jù)生成管道

2

存儲和分區(qū)

通常，在構(gòu)建一個需要存儲層的回放系統(tǒng)時，人們可能會使用基于Hadoop和HDFS的架構(gòu)。但我們選擇了Kafka，主要基于以下兩個原因：

·已有的實時系統(tǒng)采用了發(fā)布和訂閱架構(gòu)；

·回放系統(tǒng)存儲的事件量不是PB級的，我們存儲的數(shù)據(jù)不會超過幾天。此外，執(zhí)行Hadoop的MapReduce作業(yè)比在Kafka上消費數(shù)據(jù)成本更高、速度更慢，達不到開發(fā)者的期望。

要利用實時管道來構(gòu)建回放管道，首先要確保事件被存儲在Kafka中。我們把Kafka主題叫作deliverylog，每個數(shù)據(jù)中心都有一個這樣的主題。然后，這些主題被交叉復(fù)制，實現(xiàn)數(shù)據(jù)冗余，以便支持來自數(shù)據(jù)中心之外的重放請求。事件在被傳遞之前經(jīng)過去重操作。

在這個Kafka主題上，我們使用默認的分區(qū)機制創(chuàng)建了多個分區(qū)，分區(qū)與WebhookId的散列值（事件記錄的鍵）一一對應(yīng)。我們考慮過使用靜態(tài)分區(qū)，但最終決定不使用它，因為如果其中一個開發(fā)人員生成的事件多于其他開發(fā)人員，那么這個分區(qū)包含的數(shù)據(jù)將多于其他分區(qū)，造成了分區(qū)的不均衡。相反，我們選擇固定數(shù)量的分區(qū)，然后使用默認分區(qū)策略來分布數(shù)據(jù)，這樣就降低了分區(qū)不均衡的風(fēng)險，并且不需要讀取Kafka主題的所有分區(qū)。重放服務(wù)基于請求的WebhookId來確定要讀取哪個分區(qū)，并為該分區(qū)啟動一個新的Kafka消費者。主題的分區(qū)數(shù)量不會發(fā)生變化，因為這會影響鍵的散列和事件的分布。

我們使用了固態(tài)磁盤，根據(jù)每個時間段讀取的事件數(shù)量來分配空間。我們選擇這種磁盤而不是傳統(tǒng)的硬盤驅(qū)動器，以此來獲得更快的處理速度，并減少與查找和訪問操作相關(guān)的開銷。因為我們需要訪問低頻數(shù)據(jù)，無法獲得頁面緩存優(yōu)化的好處，所以最好是使用固態(tài)磁盤。

為了最小化存儲空間，我們使用了snappy壓縮算法。我們知道大部分處理工作都在消費端，之所以選擇snappy，是因為它在解壓時比其他Kafka所支持的壓縮算法(如gzip和lz4)更快。

3

請求和處理

在我們設(shè)計的這個系統(tǒng)中，通過API調(diào)用來發(fā)快遞重放請求。我們從請求消息體中獲取WebhookId和要重放的事件的日期范圍。這些請求被持久化到MySQL中，相當(dāng)于進入了隊列，直到它們被重放服務(wù)讀取。請求中的日期范圍用于確定要讀取的分區(qū)的偏移量。消費者對象的offsetForTimes函數(shù)用于獲取偏移量。

圖2：重放系統(tǒng)接收請求，并將請求發(fā)快遞給配置服務(wù)（數(shù)據(jù)訪問層），然后被持久化到數(shù)據(jù)庫中。

重放服務(wù)處理每個重放請求，它們通過MySQL相互協(xié)調(diào)，處理數(shù)據(jù)庫中的下一個需要重放的記錄。重放進程定期輪詢MySQL，獲取需要被處理的掛起作業(yè)。一個請求會在各種狀態(tài)之間轉(zhuǎn)換。等待被處理的請求處于開放狀態(tài)（OPEN STATE），剛退出隊列的請求處于啟動狀態(tài)（STARTED STATE），正在被處理的請求處于進行中狀態(tài)（ONGOING STATE），已處理完成的請求將轉(zhuǎn)換到已完成狀態(tài)（COMPLETED STATE）。一個重放進程只會選擇一個尚未啟動的請求(即處于打開狀態(tài)的請求)。

每隔一段時間，當(dāng)一個工作進程將一個請求退出隊列后，它會在MySQL表中寫入時間戳，表示正在處理當(dāng)前的重放作業(yè)。如果重放進程在處理請求時死掉了，相應(yīng)的作業(yè)將被重新啟動。因此，除了將處于打開狀態(tài)的請求退出隊列之外，重放操作還將讀取處于已開始或正在進行中的、在預(yù)定義的分鐘數(shù)內(nèi)沒有心跳的作業(yè)。

圖3：數(shù)據(jù)傳遞層：重放服務(wù)通過輪詢MySQL來讀取作業(yè)，消費來自Kafka的消息，并通過Webhook服務(wù)傳遞事件。

在讀取事件時會進行去重操作，然后事件被發(fā)布到消費者端的Webhook URL上。去重是通過維護被讀取事件的散列值緩存來實現(xiàn)的。如果遇到具有相同散列值的事件，就不傳遞這個事件。

總的來說，我們的解決方案與傳統(tǒng)的將Kafka作為實時、流式系統(tǒng)的用法不一樣。我們成功地將Kafka作為存儲系統(tǒng)，構(gòu)建了一個API，在進行事件恢復(fù)時提升了用戶體驗和數(shù)據(jù)訪問能力。我們利用已有的實時系統(tǒng)設(shè)計讓系統(tǒng)的維護變得更加容易。此外，客戶數(shù)據(jù)的恢復(fù)速度達到了我們的預(yù)期。

原文鏈接

https://blog.twitter.com/engineering/enus/topics/infrastructure/2020/kafkaasastoragesystem.html

特別聲明：以上文章內(nèi)容僅代表作者本人觀點，不代表ESG跨境電商觀點或立場。如有關(guān)于作品內(nèi)容、版權(quán)或其它問題請于作品發(fā)表后的30日內(nèi)與ESG跨境電商聯(lián)系。