《智能運維實踐 蘇娜 孫琳 王鴿著 人工智能技術叢書 自然語言處理的常用算法 日志異常檢測 根因定位 網絡流量異常檢測 清華大學出版社》【摘要 書評 試讀】- 京東圖書

數據轉換是數據預處理中的關鍵步驟，用于將數據從原始格式轉換為適合分析和建模的形式。這一過程能夠提高數據質量，確保不同來源或格式的數據在分析時具有一致性和可比性。數據轉換通常包括多種技術和方法，例如單位轉換、標準化、格式化、歸一化以及特征編碼等。下面以傳感器數據為例，詳細介紹數據轉換的方法和過程，如示例6-4所示，示例文件為demo/code/chapter6/Data_transformation.py。

1. 示例背景

運維團隊需要處理來自不同系統的監控數據，這些數據可能存在以下問題：

• 單位不統一：存儲容量單位有GB、MB、KB，時間單位有毫秒（ms）、秒（s）、納秒（ns）。
• 格式不一致：時間戳格式有UNIX時間戳、ISO8601等多種格式。
• 標準化缺失：狀態碼存在數字（如200）和文本（如"success"）混用的情況。

2. 數據轉換

具體數據轉換如下：

• 單位轉換：將所有容量統一為GB，時間統一為秒。
• 時間標準化：轉換為UTC時區的ISO8601格式。
• 狀態碼規范化：統一為HTTP狀態碼數字。
• 數據格式化：輸出為結構化的JSON格式文件。

【示例6-4】數據轉換

import re
import json
from datetime import datetime
import pytz
from enum import Enum# 示例原始數據
raw_data = ['{"host": "server1", "disk_usage": "500GB", "response_time": 
120ms", "timestamp": 1678923456, "status": "success"}','{"host": "server2", "disk_usage": "250000MB", "response_time": 
0.45", "timestamp": "2023-03-15T12:30:45+08:00", "status": 200}','{"host": "server3", "disk_usage": "1000000KB", "response_time": 
1.2s", "timestamp": "15/Mar/2023:12:30:45", "status": "error"}'
]class StatusCode(Enum):SUCCESS = 200ERROR = 500UNKNOWN = 0def convert_size_to_gb(size_str):"""將存儲容量統一轉換為GB"""size_str = str(size_str).upper()if 'GB' in size_str:return float(size_str.replace('GB', ''))elif 'MB' in size_str:return float(size_str.replace('MB', '')) / 1024elif 'KB' in size_str:return float(size_str.replace('KB', '')) / (1024 * 1024)else:  # 無單位假定為GBreturn float(size_str)def convert_time_to_seconds(time_str):"""將時間統一轉換為秒"""time_str = str(time_str).lower()if 'ms' in time_str:return float(time_str.replace('ms', '')) / 1000elif 's' in time_str:return float(time_str.replace('s', ''))elif 'ns' in time_str:return float(time_str.replace('ns', '')) / 1e9else:  # 無單位假定為秒return float(time_str)def standardize_timestamp(timestamp):"""標準化時間戳為UTC ISO8601格式"""if isinstance(timestamp, (int, float)):return datetime.fromtimestamp(timestamp, pytz.utc).isoformat()elif 'T' in str(timestamp):dt = datetime.fromisoformat(str(timestamp))if dt.tzinfo is None:dt = pytz.utc.localize(dt)else:dt = dt.astimezone(pytz.utc)return dt.isoformat()elif '/' in str(timestamp):dt = datetime.strptime(str(timestamp), '%d/%b/%Y:%H:%M:%S')dt = pytz.timezone('Asia/Shanghai').localize(dt). astimezone(pytz.utc)return dt.isoformat()return Nonedef standardize_status(status):"""標準化狀態碼"""status = str(status).lower()if status in ('success', 'ok', '200'):return StatusCode.SUCCESS.valueelif status in ('error', 'fail', '500'):return StatusCode.ERROR.valueelif status.isdigit():return int(status)return StatusCode.UNKNOWN.valuedef process_data_entry(entry):"""處理單個數據條目"""try:data = json.loads(entry)except json.JSONDecodeError:print(f"無效JSON格式: {entry}")return Nonereturn {'host': data['host'],'disk_usage_gb': convert_size_to_gb(data['disk_usage']),'response_time_seconds': 
onvert_time_to_seconds(data['response_time']),'timestamp_utc': standardize_timestamp(data['timestamp']),'status_code': standardize_status(data['status'])}def analyze_processed_data(processed_data):"""分析處理后的數據"""valid_data = [x for x in processed_data if x is not None]disk_usages = [x['disk_usage_gb'] for x in valid_data]response_times = [x['response_time_seconds'] for x in valid_data]status_codes = [x['status_code'] for x in valid_data]return {'total_entries': len(processed_data),'valid_entries': len(valid_data),'avg_disk_usage': sum(disk_usages) / len(disk_usages) if disk_usages else None,'avg_response_time': sum(response_times) / len(response_times) if response_times else None,'status_distribution': {code: status_codes.count(code) for code in set(status_codes)}}def main():print("原始數據示例:")for data in raw_data[:2]:print(data)print("\n正在處理數據...")processed_data = [process_data_entry(entry) for entry in raw_data]print("\n標準化后的數據示例:")for data in processed_data[:2]:print(json.dumps(data, indent=2))# 分析數據analysis = analyze_processed_data(processed_data)print("\n數據分析結果:")print(json.dumps(analysis, indent=2))# 保存結果with open('data\processed_data.json', 'w') as f:json.dump(processed_data, f, indent=2)print("\n結果已保存到 data\processed_data.json")if __name__ == "__main__":main()

代碼解釋：

1）單位轉換函數

• convert_size_to_gb(): 將存儲容量單位從GB、MB、KB統一轉換為GB。
• convert_time_to_seconds(): 將時間單位從毫秒（ms）、秒（s）、納秒（ns）統一轉換為秒。

2）標準化函數

• standardize_timestamp(): 處理多種時間戳格式，并統一為UTC時區的ISO8601格式。
• standardize_status(): 統一狀態碼表示，使用枚舉規范狀態碼。

3）數據處理流程

• 使用process_data_entry()處理單個數據條目，應用所有轉換規則。
• 保留原始數據以便追溯。
• 簡化處理邏輯，不處理缺失值情況。

4）數據分析

• 計算平均磁盤使用量和響應時間。
• 統計狀態碼分布。
• 輸出數據質量報告。

運行示例6-4，處理結果保存至.\data\processed_data.json，控制臺輸出如下：

原始數據示例:
{"host": "server1", "disk_usage": "500GB", "response_time": "120ms", "timestamp": 1678923456, "status": "success"}
{"host": "server2", "disk_usage": "250000MB", "response_time": "0.45", "timestamp": "2023-03-15T12:30:45+08:00", "status": 200}正在處理數據...標準化后的數據示例:
{"host": "server1","disk_usage_gb": 500.0,"response_time_seconds": 0.12,"timestamp_utc": "2023-03-15T23:37:36+00:00","status_code": 200
}
{"host": "server2","disk_usage_gb": 244.140625,"response_time_seconds": 0.45,"timestamp_utc": "2023-03-15T04:30:45+00:00","status_code": 200
}數據分析結果:
{"total_entries": 3,"valid_entries": 3,"avg_disk_usage": 248.3647664388021,"avg_response_time": 0.59,"status_distribution": {"200": 2,"500": 1}
}結果已保存到 data\processed_data.json