如何在Python中不结合旧数据更新孤立算法模型并保留旧知识

技术选型 #

• 在线孤立森林：使用river（原creme）库的IsolationForest，支持流式增量更新，无需保留旧数据或重新训练全量样本，仅用新数据更新模型同时保留原有知识。
• 日志自动获取：用watchdog监控日志文件/目录变化，触发DSL提取新数据；或用schedule定时执行DSL查询拉取新增日志。

完整Python代码实现 #

1. 依赖安装 #

pip install river watchdog elasticsearch schedule

2. 日志监控与数据提取模块 #

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from elasticsearch import Elasticsearch
import pandas as pd
import datetime

# 初始化ES客户端（根据实际配置修改）
es = Elasticsearch(["http://your-es-host:9200"])

def extract_new_logs(last_processed_time):
    """用DSL查询获取新增网络日志（示例DSL，需匹配实际日志结构调整）"""
    dsl_query = {
        "query": {
            "range": {
                "@timestamp": {
                    "gt": last_processed_time
                }
            }
        },
        "_source": ["src_ip", "dest_port", "bytes_transferred", "request_time"]  # 替换为你的模型特征字段
    }
    response = es.search(index="network-logs", body=dsl_query, size=1000)
    hits = response["hits"]["hits"]
    if not hits:
        return pd.DataFrame(), last_processed_time
    # 转换为DataFrame格式
    data = pd.DataFrame([hit["_source"] for hit in hits])
    # 更新最后处理时间为最新日志的时间戳
    new_last_time = max([hit["_source"]["@timestamp"] for hit in hits])
    return data, new_last_time

class LogFileHandler(FileSystemEventHandler):
    def __init__(self, model, last_time):
        self.model = model
        self.last_processed_time = last_time

    def on_created(self, event):
        """新日志文件创建时触发数据提取与模型更新"""
        if not event.is_directory:
            print(f"检测到新日志文件：{event.src_path}")
            new_data, self.last_processed_time = extract_new_logs(self.last_processed_time)
            if not new_data.empty:
                # 逐样本增量更新模型
                for _, row in new_data.iterrows():
                    sample = row.to_dict()
                    self.model.learn_one(sample)
                print(f"已用{len(new_data)}条新日志完成模型更新")

3. 模型初始化与监控启动 #

from river.anomaly import IsolationForest
import time

if __name__ == "__main__":
    # 初始化在线孤立森林模型（参数可根据业务需求调整）
    model = IsolationForest(n_trees=100, window_size=200)
    # 初始最后处理时间（可从本地存储读取上次结束时间，此处用当前时间）
    last_processed_time = datetime.datetime.now().isoformat()

    # 监控目标日志目录（替换为你的网络日志存储路径）
    log_dir = "/var/log/network"
    event_handler = LogFileHandler(model, last_processed_time)
    observer = Observer()
    observer.schedule(event_handler, log_dir, recursive=False)
    observer.start()

    print(f"开始监控日志目录：{log_dir}")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

4. 定时拉取替代方案（适用于集中存储的日志） #

如果日志存储在数据库/ES而非本地文件系统，可改用定时任务拉取新增数据：

import schedule

def update_model_job(model, last_time):
    new_data, updated_last_time = extract_new_logs(last_time)
    if not new_data.empty:
        for _, row in new_data.iterrows():
            model.learn_one(row.to_dict())
        print(f"定时任务完成：已用{len(new_data)}条新日志更新模型")
    return updated_last_time

if __name__ == "__main__":
    model = IsolationForest(n_trees=100, window_size=200)
    last_processed_time = datetime.datetime.now().isoformat()

    # 每5分钟执行一次拉取与更新
    schedule.every(5).minutes.do(lambda: update_model_job(model, last_processed_time))

    while True:
        schedule.run_pending()
        time.sleep(1)

关键实现说明 #

• 增量更新逻辑：river的IsolationForest采用在线学习机制，每棵树会根据新样本动态调整分裂节点，无需重新训练全部树结构，大幅降低计算开销。
• 旧知识保留：模型通过维护树的层级结构与分裂阈值，自动保留从旧数据中学到的正常模式，新数据仅补充调整对新异常的识别能力。
• 数据获取灵活性：可根据日志存储方式选择文件监控或定时拉取，DSL查询需匹配你的日志字段结构做针对性调整。

内容的提问来源于stack exchange，提问作者user19000614