前情回顧：

將 AzureBlob 的日志通過 Azure Event Hubs 發(fā)給 Elasticsearch（1）-CSDN博客

前邊的方案是挺好的，但?Azure Event Hubs 是付費(fèi)服務(wù)，我這里只是一個(gè)獲取日志進(jìn)行必要的分析，并且不要求實(shí)時(shí)性，關(guān)鍵?沒錢，那怎么辦呢？

替代方案：

如果對(duì)實(shí)時(shí)性沒有嚴(yán)格要求，并且希望避免使用付費(fèi)的 Azure Event Hubs 服務(wù)，可以采用更經(jīng)濟(jì)的方式，比如將 Azure Blob 存儲(chǔ)的日志直接發(fā)送到 Elasticsearch 或通過 Azure Storage Queue 和 Azure Function 來實(shí)現(xiàn)。這些方法可以有效減少成本，同時(shí)滿足日志解析和分析需求。

1. 直接從 Azure Blob 存儲(chǔ)讀取日志并處理

思路

直接從 Azure Blob 存儲(chǔ) 中讀取生成的日志文件，解析所需字段后發(fā)送到 Elasticsearch。這種方式適合沒有實(shí)時(shí)性需求的場(chǎng)景，定期運(yùn)行任務(wù)即可。

實(shí)現(xiàn)步驟

1. 啟用 Azure 存儲(chǔ)日志記錄：

   • 在存儲(chǔ)賬戶中啟用 診斷設(shè)置。
   • 將日志記錄輸出到同一存儲(chǔ)賬戶的 Blob 容器中，例如 insights-logs。

2. 使用 Azure Function 或定時(shí)任務(wù)讀取日志：

   • 編寫腳本（可以用 Python、PowerShell、C# 等語言），從指定的 Blob 容器中下載日志文件。
   • 解析日志文件，根據(jù)需要提取字段并格式化為 Elasticsearch 可接受的 JSON 數(shù)據(jù)。
   • 將數(shù)據(jù)批量發(fā)送到 Elasticsearch。

3. 示例代碼：

   • 使用 Python 和 azure-storage-blob 庫：

   from azure.storage.blob import BlobServiceClient
   from elasticsearch import Elasticsearch
   import json# Azure Blob Storage 配置
   storage_account_name = "your_storage_account"
   storage_account_key = "your_storage_account_key"
   container_name = "insights-logs"# Elasticsearch 配置
   es = Elasticsearch("http://your-elasticsearch-server:9200")def process_blob_logs():blob_service_client = BlobServiceClient(account_url=f"https://{storage_account_name}.blob.core.windows.net",credential=storage_account_key)container_client = blob_service_client.get_container_client(container_name)# 列出日志文件blobs = container_client.list_blobs()for blob in blobs:blob_client = container_client.get_blob_client(blob)log_data = blob_client.download_blob().readall().decode('utf-8')# 假設(shè)日志是 JSON 格式for line in log_data.splitlines():try:log_entry = json.loads(line)# 提取你需要的字段parsed_entry = {"timestamp": log_entry["time"],"operation": log_entry["operationName"],"blobName": log_entry.get("blobName"),"requestorIp": log_entry.get("requestorIpAddress"),}# 寫入 Elasticsearches.index(index="storage-logs", document=parsed_entry)except Exception as e:print(f"Error processing log entry: {e}")if __name__ == "__main__":process_blob_logs()
   

4. 設(shè)置定時(shí)運(yùn)行：

   • 如果使用 Azure Function，配置 Timer Trigger 定期運(yùn)行該腳本。
   • 如果使用本地腳本，使用 cron（Linux）或計(jì)劃任務(wù)（Windows）實(shí)現(xiàn)定時(shí)任務(wù)。

---

2. 使用 Azure Storage Queue

思路

利用 Azure Storage Queue 作為消息隊(duì)列替代 Event Hubs，用于暫存日志文件元數(shù)據(jù)或小型消息，然后由處理程序（如 Azure Function 或腳本）消費(fèi)隊(duì)列消息，讀取并處理日志。

實(shí)現(xiàn)步驟

1. 啟用日志記錄：

   • 將日志寫入 Blob 存儲(chǔ)。

2. 配置 Azure Storage Queue：

   • 創(chuàng)建一個(gè) Azure Queue。
   • 編寫腳本將日志的元數(shù)據(jù)（例如 Blob 的路徑）寫入隊(duì)列。

3. 編寫處理腳本：

   • 消費(fèi)隊(duì)列消息，根據(jù)消息中的 Blob 路徑讀取并解析日志文件。

4. 示例代碼：

   • 使用 Python 和 azure-storage-queue：

   from azure.storage.queue import QueueClient
   from azure.storage.blob import BlobServiceClient
   import jsonqueue_name = "your-queue"
   storage_account_name = "your_storage_account"
   storage_account_key = "your_storage_account_key"def process_queue_messages():queue_client = QueueClient(account_url=f"https://{storage_account_name}.queue.core.windows.net",credential=storage_account_key,queue_name=queue_name)blob_service_client = BlobServiceClient(account_url=f"https://{storage_account_name}.blob.core.windows.net",credential=storage_account_key)messages = queue_client.receive_messages()for msg in messages:blob_path = msg.content ?# 假設(shè)隊(duì)列消息中存儲(chǔ)的是 Blob 路徑container_name, blob_name = blob_path.split('/', 1)blob_client = blob_service_client.get_blob_client(container_name, blob_name)log_data = blob_client.download_blob().readall().decode('utf-8')# 解析日志并發(fā)送到 Elasticsearch（同上）print(f"Processing blob: {blob_path}")queue_client.delete_message(msg) ?# 刪除已處理消息if __name__ == "__main__":process_queue_messages()
   

---

3. 直接讀取和解析日志文件后推送到 Elasticsearch

這種方法可以完全避開隊(duì)列服務(wù)，直接通過腳本定期從存儲(chǔ)賬戶下載日志文件并解析后推送到 Elasticsearch。

注意事項(xiàng)

• 定時(shí)任務(wù)頻率：根據(jù)日志生成的頻率設(shè)定腳本運(yùn)行的時(shí)間間隔。
• 日志存儲(chǔ)策略：Blob 日志文件可能會(huì)快速增長，考慮啟用存儲(chǔ)生命周期管理規(guī)則定期刪除過期日志文件。
• 安全性：確保存儲(chǔ)帳戶密鑰或連接字符串的安全性，可使用 Azure 的 Managed Identity 替代密鑰。

---

比較與選擇

方案	適用場(chǎng)景	實(shí)現(xiàn)復(fù)雜度	成本
直接從 Blob 存儲(chǔ)讀取	數(shù)據(jù)量不大，無實(shí)時(shí)性需求	低	低
使用 Storage Queue	有一定的隊(duì)列需求，異步處理	中	較低
使用 Azure Event Hubs	需要高吞吐量和實(shí)時(shí)性	高	高

推薦：

• 如果數(shù)據(jù)量不大，選擇 直接從 Blob 存儲(chǔ)讀取。
• 如果需要解耦消息和處理程序，可以選擇 Azure Storage Queue。

---

前情后續(xù)：

將 AzureBlob 的日志通過 Azure Event Hubs 發(fā)給 Elasticsearch（1）-CSDN博客

將 AzureBlob 的日志通過 Azure Event Hubs 發(fā)給 Elasticsearch（2）-CSDN博客

將 AzureBlob 的日志通過 Azure Event Hubs 發(fā)給 Elasticsearch（3）-CSDN博客

---

---

---