整合 Airflow 記錄與 Azure 監視器

文章
06/13/2024

您將在本文中，了解如何開始將適用於能源產業的 Microsoft Azure 資料管理員執行個體的 Airflow 記錄收集到 Azure 監視器中。這項整合功能可協助您偵錯 Airflow DAG (有向非循環圖) 執行失敗。

必要條件

現有的 Log Analytics 工作區。此工作區將用來查詢 Airflow 記錄，使用 Log Analytics 工作區中的 Kusto 查詢語言 (KQL) 查詢編輯器。實用資源：在 Azure 入口網站中建立記錄分析工作區。
現有的儲存體帳戶：將用來儲存 Airflow 記錄的 JSON 傾印。儲存體帳戶不一定與 Log Analytics 工作區位於相同的訂用帳戶中。

啟用儲存體帳戶中的診斷設定以收集記錄

每個適用於能源的 Azure Data Manager 實例都會內建 Azure Data Factory 工作流程協調流程管理員（由 Apache Airflow 提供）實例。我們會收集 Airflow 記錄以進行內部疑難排解和偵錯。 Airflow 記錄可以透過下列方式與 Azure 監視器整合：

儲存體帳戶
Log Analytics 工作區

若要透過上述兩個選項之一來存取記錄，您需要建立診斷設定。每個診斷設定都有三個基本部分：

部分	描述
Name	這是診斷記錄的名稱。請務必為每個記錄設定唯一的名稱。
類別	要傳送到每個目的地的記錄類別。每個 Azure 服務的類別集合都會有所不同。瀏覽：支援的資源記錄類別
目的地	一個或多個要傳送記錄的目的地。所有 Azure 服務都會共用一組相同的可能目的地。每個診斷設定可以定義一個或多個目的地，但不能有多個特定類型的目的地。類型應為儲存體帳戶、事件中樞命名空間或事件中樞。

請遵循下列步驟來進行診斷設定：

開啟適用於能源產業的 Microsoft Azure 資料管理員的 [概觀] 頁面
從左側面板中選取 [診斷設定]
選取 [新增診斷設定]
選取 [記錄] 底下的 [Airflow 工作記錄]
選取 [封存至儲存體帳戶]
確認您要封存記錄的訂用帳戶和儲存體帳戶。

瀏覽儲存體帳戶以下載 Airflow 記錄

建立診斷設定以將 Airflow 工作記錄封存到儲存體帳戶之後，您可以瀏覽至儲存體帳戶 [概觀] 頁面。然後，您可以使用左側面板中的 [儲存體瀏覽器] 來尋找您想要調查的適當 JSON 檔案。您可以在年、月、日之間移動，直覺瀏覽不同的目錄。

透過左側面板可以瀏覽 [容器]。
開啟右側的資訊窗格。其中包含 [下載] 按鈕，可在本機儲存記錄檔。
您可以在任何編輯器中分析下載的記錄。

啟用診斷設定以將記錄整合至 Log Analytics 工作區

您可以在適用於能源產業的 Microsoft Azure 資料管理員執行個體概觀頁面上，使用左側面板下的 [診斷設定]，將 Airflow 記錄與 Log Analytics 工作區整合。

在 Log Analytics 工作區中使用整合的 Airflow 記錄

使用 Kusto 查詢語言 (KQL) ，從 Log Analytics 工作區擷取已收集 Airflow 記錄上的所需資料。您可以將預先建置的範例查詢載入 Log Analytics 工作區，或建立您自己的查詢。

載入預先建置的查詢：從資源的功能表中選取 [記錄]。 Log Analytics 隨即開啟，其中的 [查詢] 視窗包含您資源類型的預先建置查詢。瀏覽可用的查詢。找出所需的項目，並選取 [執行]。查詢會新增至查詢視窗，並傳回結果。
在查詢編輯器中撰寫查詢：您可以在 Log Analytics 工作區的查詢編輯器中複製、貼上和編輯下列查詢，或以 KQL 自行撰寫。

範例查詢

此查詢會傳回層級為 ERROR 的所有 Airflow 記錄。您可以為適用於能源產業的 Azure 資料管理員新增 (取消註解) where 子句和 DAG 執行的相互關聯識別碼來篩選結果。


OEPAirFlowTask
| extend ResourceName = tostring(split(_ResourceId , '/')[-1])
// | where ResourceName == "<the name of ADME instance>"        // to filter on resourceName replace <...> and uncomment line
// | where CorrelationId == "<DAG run's runId>"                 // to filter on correlationID replace <...> with correlationId (same as runId) - we have created a duplicate for to maintain consistency of column name across all services 
| where LogLevel  == "ERROR"
| project TimeGenerated, DagName, LogLevel, DagTaskName, CodePath, Content

此查詢會列出所有 DAG 執行，以及上述適用於能源產業的 Azure 資料管理員資源中這些 DAG 執行對應的相互關聯識別碼。


OEPAirFlowTask
| extend ResourceName = tostring(split(_ResourceId , '/')[-1])
// | where ResourceName == "<the name of ADME instance>"        // to filter on resourceName replace <...> and uncomment line
| distinct DagName, CorrelationId                               // correlationId is same as runId - we have created a duplicate for consistency in search across logs of all services 
| sort by DagName asc

我們已新增文件，以協助您使用 Log Analytics 工作區中收集的 Airflow 記錄，對資訊清單擷取流程進行疑難排解。

下一步

現在系統已在收集資源記錄，您可建立記錄查詢警示，以便在記錄資料中識別出有趣的資料時主動收到通知。

建立 Azure 資源的記錄查詢警示

分享方式：