效能微調 - 事件串流

Azure Functions

Azure IoT 中樞

Azure Cosmos DB

本文說明開發小組如何使用計量來尋找瓶頸，並改善分散式系統的效能。本文是以我們針對範例應用程式所做的實際負載測試為基礎。

本文是系列文章的其中一篇。請在這裡閱讀第一個部分。

案例：使用Azure Functions處理事件串流。

事件串流架構的圖表

在此案例中，無人機團隊會即時傳送位置資料給Azure IoT 中樞。 Functions 應用程式會接收事件、將資料轉換成 GeoJSON 格式，並將轉換的資料寫入 Azure Cosmos DB。 Azure Cosmos DB 具有地理空間資料的原生支援，而 Azure Cosmos DB 集合可以編制索引，以便有效率的空間查詢。例如，用戶端應用程式可以在指定位置的 1 公里內查詢所有無人機，或在特定區域內尋找所有無人機。

這些處理需求很簡單，不需要完整串流處理引擎。特別是，處理不會聯結資料流程、匯總資料或跨時間範圍處理。根據這些需求，Azure Functions非常適合用來處理訊息。 Azure Cosmos DB 也可以調整以支援非常高的寫入輸送量。

監視輸送量

此案例提供有趣的效能挑戰。 已知每個裝置的資料速率，但裝置數目可能會變動。在此商務案例中，延遲需求並不特別嚴格。無人機的報告位置只需要在一分鐘內正確無誤。也就是說，函式應用程式必須跟上一段時間內的平均擷取速率。

IoT 中樞會將訊息儲存在記錄資料流程中。傳入訊息會附加至資料流程的結尾。資料流程的讀取器，在此案例中，函式應用程式會控制自己的周遊資料流程速率。這種讀取和寫入路徑分離會讓IoT 中樞非常有效率，但也表示讀取器可能會落後。為了偵測此狀況，開發小組新增了自訂計量來測量訊息延遲。此計量會記錄訊息到達IoT 中樞，以及函式收到訊息進行處理時之間的差異。

var ticksUTCNow = DateTimeOffset.UtcNow;

// Track whether messages are arriving at the function late.
DateTime? firstMsgEnqueuedTicksUtc = messages[0]?.EnqueuedTimeUtc;
if (firstMsgEnqueuedTicksUtc.HasValue)
{
    CustomTelemetry.TrackMetric(
                        context,
                        "IoTHubMessagesReceivedFreshnessMsec",
                        (ticksUTCNow - firstMsgEnqueuedTicksUtc.Value).TotalMilliseconds);
}

方法 TrackMetric 會將自訂計量寫入 Application Insights。如需在 Azure 函式內使用 TrackMetric 的相關資訊，請參閱 C# 函式中的自訂遙測。

如果函式持續掌握訊息數量，此計量應保持低穩定狀態。有些延遲是無法避免的，因此值永遠不會是零。但是，如果函式落後，排入佇列的時間和處理時間之間的差異將會開始增加。

測試 1：基準

第一個負載測試顯示立即問題：函式應用程式一致收到來自 Azure Cosmos DB 的 HTTP 429 錯誤，指出 Azure Cosmos DB 正在節流寫入要求。

Azure Cosmos DB 節流要求的圖表

為了回應，小組藉由增加為集合配置的 RU 數目來調整 Azure Cosmos DB，但會繼續發生錯誤。這似乎很奇怪，因為其信封後計算顯示 Azure Cosmos DB 應該不會因寫入要求量而發生問題。

該日後，其中一位開發人員將下列電子郵件傳送給小組：

我查看了 Azure Cosmos DB 中的暖路徑。我沒有了解的一件事。分割區索引鍵是 deliveryId，不過我們不會將 deliveryId 傳送至 Azure Cosmos DB。我是否遺漏某些專案？

這是線索。查看分割區熱度圖，結果顯示所有檔都登陸在相同的分割區上。

Azure Cosmos DB 分割區熱度圖的圖表

您想要在熱度圖中看到的內容是所有資料分割的偶數分佈。在此情況下，因為每個檔都已寫入相同的分割區，所以新增 RU 沒有説明。問題已變成程式碼中的 Bug。雖然 Azure Cosmos DB 集合具有分割區索引鍵，但 Azure 函式實際上並未在檔中包含資料分割索引鍵。如需分割區熱度圖的詳細資訊，請參閱判斷分割區之間的輸送量分佈。

測試 2：修正資料分割問題

當小組部署程式碼修正並重新執行測試時，Azure Cosmos DB 會停止節流。一段時間，一切看起來都不錯。但在特定的負載中，遙測顯示函式撰寫的檔應該較少。下圖顯示從IoT 中樞接收的訊息，以及寫入 Azure Cosmos DB 的檔。黃色行是每個批次收到的訊息數目，而綠色則是每個批次寫入的檔數目。這些應該是成正比的。相反地，每個批次的資料庫寫入作業數目會大幅下降到大約 07：30。

已捨棄訊息的圖表

下一個圖表顯示訊息從裝置抵達IoT 中樞，以及函式應用程式處理該訊息時之間的延遲。您可以在相同的時間點看到，延遲尖峰大幅增加、關機和下降。

訊息延遲的圖表

值在 5 分鐘尖峰，然後下降至零的原因是函式應用程式會捨棄延遲超過 5 分鐘的訊息：

foreach (var message in messages)
{
    // Drop stale messages,
    if (message.EnqueuedTimeUtc < cutoffTime)
    {
        log.Info($"Dropping late message batch. Enqueued time = {message.EnqueuedTimeUtc}, Cutoff = {cutoffTime}");
        droppedMessages++;
        continue;
    }
}

當延遲計量回復為零時，您可以在圖表中看到此情況。同時，資料已遺失，因為函式已擲回訊息。

發生什麼事？針對這個特定的負載測試，Azure Cosmos DB 集合具有可備援的 RU，因此瓶頸不在資料庫上。相反地，問題是在訊息處理迴圈中。簡單來說，函式並未快速撰寫檔，以跟上傳入的訊息量。經過一段時間後，它會進一步和更進一步地落後。

測試 3：平行寫入

如果處理訊息的時間是瓶頸，其中一個解決方案是平行處理更多訊息。在此情節中：

增加IoT 中樞分割區的數目。每個IoT 中樞分割區都會一次指派一個函式實例，因此我們預期輸送量會隨著分割區數目以線性方式調整。
平行處理函式內的檔寫入。

為了探索第二個選項，小組修改了函式以支援平行寫入。函式的原始版本使用 Azure Cosmos DB 輸出系結。優化版本會直接呼叫 Azure Cosmos DB 用戶端，並使用 Task.WhenAll平行執行寫入：

private async Task<(long documentsUpserted,
                    long droppedMessages,
                    long cosmosDbTotalMilliseconds)>
                ProcessMessagesFromEventHub(
                    int taskCount,
                    int numberOfDocumentsToUpsertPerTask,
                    EventData[] messages,
                    TraceWriter log)
{
    DateTimeOffset cutoffTime = DateTimeOffset.UtcNow.AddMinutes(-5);

    var tasks = new List<Task>();

    for (var i = 0; i < taskCount; i++)
    {
        var docsToUpsert = messages
                            .Skip(i * numberOfDocumentsToUpsertPerTask)
                            .Take(numberOfDocumentsToUpsertPerTask);
        // client will attempt to create connections to the data
        // nodes on Azure Cosmos DB clusters on a range of port numbers
        tasks.Add(UpsertDocuments(i, docsToUpsert, cutoffTime, log));
    }

    await Task.WhenAll(tasks);

    return (this.UpsertedDocuments,
            this.DroppedMessages,
            this.CosmosDbTotalMilliseconds);
}

請注意，使用方法可能會有競爭條件。假設來自相同無人機的兩則訊息都發生在相同的訊息批次中。藉由平行寫入它們，先前的訊息可能會覆寫稍後的訊息。在此特定案例中，應用程式可以容許偶爾遺失訊息。無人機每隔 5 秒傳送新的位置資料，因此 Azure Cosmos DB 中的資料會持續更新。不過，在其他案例中，嚴格地依序處理訊息可能很重要。

部署此程式碼變更之後，應用程式就能夠使用具有 32 個分割區的IoT 中樞擷取超過 2500 個要求/秒。

用戶端考慮

整體用戶端體驗可能會因伺服器端上的積極平行處理而降低。請考慮使用此實作中未顯示 azure Cosmos DB 大量執行程式程式庫 () 這可大幅減少將配置給 Azure Cosmos DB 容器之輸送量飽和所需的用戶端計算資源。相較于多執行緒應用程式，使用大量匯入 API 寫入資料的單一線程應用程式，相較于在用戶端電腦的 CPU 飽和時平行寫入資料的多執行緒應用程式，可達到幾乎十倍的寫入輸送量。

摘要

在此案例中，已識別下列瓶頸：

熱寫入分割區，因為正在寫入的檔中遺漏資料分割索引鍵值。
針對每個IoT 中樞分割區以序列方式寫入檔。

為了診斷這些問題，開發小組依賴下列計量：

Azure Cosmos DB 中的節流要求。
分割區熱度圖 — 每個分割區的耗用 RU 數目上限。
收到的訊息與已建立的檔。
訊息延遲。

下一步

檢閱效能反模式

共用方式為