在 PyCharm 中建立 generate_logs.py，產生模擬資料：
# generate_logs.py
import random
import time
METHODS = ["GET", "POST", "PUT", "DELETE"]
ENDPOINTS = ["/api/v1/predict", "/api/v1/health", "/api/v1/train", "/login"]
STATUS_CODES = [200, 200, 200, 200, 404, 500, 503]
def create_dummy_log_file(filename: str = "server.log", lines: int = 100_000):
with open(filename, "w", encoding="utf-8") as f:
for _ in range(lines):
ip = f"192.168.1.{random.randint(1, 255)}"
method = random.choice(METHODS)
endpoint = random.choice(ENDPOINTS)
status = random.choice(STATUS_CODES)
latency = round(random.uniform(5.0, 1500.0), 2) # 毫秒
f.write(f"{ip} {method} {endpoint} {status} {latency}\n")
if __name__ == "__main__":
print("產生模擬 Log 檔案中...")
create_dummy_log_file()
print("產生完成！")以上程式執行後會產生 server.log 的模擬 log 檔案, 我們可以將此 dataset 拿來做資料分析, 內容列舉如下
192.168.1.171 POST /api/v1/train 500 616.19
192.168.1.129 POST /api/v1/train 503 1004.2
192.168.1.87 GET /api/v1/train 200 75.79
192.168.1.141 DELETE /api/v1/train 200 28.09「各位以前寫程式，讀檔案通常就是 readAll() 或直接開一個超大陣列把全部資料塞進去對吧？在小作業裡完全沒問題。但今天我們這門課叫『巨量資料分析』，你未來在業界要處理的是 100 GB、甚至是 1 TB 的伺服器存取記錄。你的筆電記憶體頂多 16 GB 或 32 GB，如果你企圖一次把檔案全讀進來，作業系統只會毫不客氣地送你三個英文字母：OOM (Out Of Memory)，程式直接當掉。
今天我們的目標，就是要用 Python 寫出一個**『記憶體消耗永遠只有幾 KB』**的 Log 分析器，幫往後課程的模型準備乾淨的特徵。」
今日任務
任務 1: 使用 Generator 逐行讀取，確保記憶體安全
請寫一個函數 stream_raw_logs(filepath)。
功能：傳入檔案路徑，它不要回傳一個裝滿百萬行文字的 list。
要求：使用 with open 開啟檔案，透過迴圈讀取時，不要用 return，改用 yield 把每一行文字「吐」給外面。
驗證標準：呼叫這個函數時，它應該是一個「生成器物件（generator object）」，不會一次把檔案讀完。
任務 2: 資料解析與型別轉換 (EAFP 容錯機制)
請寫一個函數 parse_log_line(line)。
輸入：單行字串，例如 "192.168.1.1 GET /api/v1/predict 200 45.2"。
輸出：解析後的結構化資料（字典 dict），包含 ip、method、endpoint、status（需為整數）、latency（需為浮點數）。
核心挑戰（容錯處理）：
如果這一行格式不對（欄位缺少、多餘，或者型別轉換失敗噴出 ValueError），不要讓程式崩潰中斷。
請用 try...except 機制捕捉異常，如果解析失敗，就回傳 None，代表這是壞掉的髒資料。
任務 3: 資料過濾與特徵匯總 (銜接 AI 特徵萃取)
請寫一個主控流程函數 process_logs(filepath)。
要求：
1. 接上任務 1 的生成器，開始讀入每一行。
2. 用任務 2 的解析函數把每一行轉成字典，遇到 None 就直接跳過（continue）。
3. 在走訪的過程中，動態更新以下統計指標：
總有效請求數（Total Requests）
平均延遲時間（Average Latency，單位毫秒）
伺服器錯誤率（Error Rate，HTTP 狀態碼 >= 400 所佔的百分比）
各 API 路徑（Endpoint）各自被呼叫了幾次
4. 最後回傳一個整理好的字典（Feature Summary），這就是未來要送進 AI 異常偵測模型的特徵向量。DEMO code, 為含有完整的 code
# log_parser.py
from typing import Generator, Dict, Any, Optional
# 任務 1: 使用 Generator 逐行讀取，確保記憶體安全
def stream_raw_logs(filepath: str) -> Generator[str, None, None]:
"""
TODO: 使用 with open 開啟檔案，並利用 yield 逐行產出文字
"""
# 任務 2: 資料解析與型別轉換 (EAFP 容錯機制)
def parse_log_line(line: str) -> Optional[Dict[str, Any]]:
"""
將單行 log 字串分割並轉為結構化字典。
格式: ip method endpoint status latency
若資料損壞或格式不符，利用 try-except 回傳 None
"""
# 任務 3: 資料過濾與特徵匯總 (銜接 AI 特徵萃取)
def process_logs(filepath: str) -> Dict[str, Any]:
raw_stream = stream_raw_logs(filepath)
# 利用推導式/生成器表達式清洗資料
parsed_stream = (parse_log_line(line) for line in raw_stream)
total_valid_requests = 0
total_latency = 0.0
error_status_count = 0
endpoint_counts: Dict[str, int] = {}
for record in parsed_stream:
if record is None:
continue
total_valid_requests += 1
total_latency += record["latency"]
if record["status"] >= 400:
error_status_count += 1
ep = record["endpoint"]
endpoint_counts[ep] = endpoint_counts.get(ep, 0) + 1
avg_latency = total_latency / total_valid_requests if total_valid_requests > 0 else 0.0
# 輸出可直接餵給後續 AI 模型（如孤立森林異常偵測）的彙整特徵
return {
"total_requests": total_valid_requests,
"avg_latency": round(avg_latency, 2),
"error_rate": round(error_status_count / total_valid_requests, 4) if total_valid_requests > 0 else 0.0,
"endpoint_distribution": endpoint_counts
}
if __name__ == "__main__":
result = process_logs("server.log")
print("=== 資料特徵彙整結果 ===")
for k, v in result.items():
print(f"{k}: {v}")
正確執行後可以看到如下結果
=== 資料特徵彙整結果 ===
total_requests: 100000
avg_latency: 753.35
error_rate: 0.4286
endpoint_distribution: {'/login': 24886, '/api/v1/health': 25116, '/api/v1/train': 25125, '/api/v1/predict': 24873}解答完整的 code
# log_parser.py
from typing import Generator, Dict, Any, Optional
# 任務 1: 使用 Generator 逐行讀取，確保記憶體安全
def stream_raw_logs(filepath: str) -> Generator[str, None, None]:
"""
TODO: 使用 with open 開啟檔案，並利用 yield 逐行產出文字
"""
with open(filepath, "r", encoding="utf-8") as file:
for line in file:
yield line.strip()
# 任務 2: 資料解析與型別轉換 (EAFP 容錯機制)
def parse_log_line(line: str) -> Optional[Dict[str, Any]]:
"""
將單行 log 字串分割並轉為結構化字典。
格式: ip method endpoint status latency
若資料損壞或格式不符，利用 try-except 回傳 None
"""
try:
parts = line.split()
if len(parts) != 5:
return None
return {
"ip": parts[0],
"method": parts[1],
"endpoint": parts[2],
"status": int(parts[3]),
"latency": float(parts[4])
}
except (ValueError, IndexError):
return None
# 任務 3: 資料過濾與特徵匯總 (銜接 AI 特徵萃取)
def process_logs(filepath: str) -> Dict[str, Any]:
raw_stream = stream_raw_logs(filepath)
# 利用推導式/生成器表達式清洗資料
parsed_stream = (parse_log_line(line) for line in raw_stream)
total_valid_requests = 0
total_latency = 0.0
error_status_count = 0
endpoint_counts: Dict[str, int] = {}
for record in parsed_stream:
if record is None:
continue
total_valid_requests += 1
total_latency += record["latency"]
if record["status"] >= 400:
error_status_count += 1
ep = record["endpoint"]
endpoint_counts[ep] = endpoint_counts.get(ep, 0) + 1
avg_latency = total_latency / total_valid_requests if total_valid_requests > 0 else 0.0
# 輸出可直接餵給後續 AI 模型（如孤立森林異常偵測）的彙整特徵
return {
"total_requests": total_valid_requests,
"avg_latency": round(avg_latency, 2),
"error_rate": round(error_status_count / total_valid_requests, 4) if total_valid_requests > 0 else 0.0,
"endpoint_distribution": endpoint_counts
}
if __name__ == "__main__":
result = process_logs("server.log")
print("=== 資料特徵彙整結果 ===")
for k, v in result.items():
print(f"{k}: {v}")