在每个特定时间间隔后训练ML模型时的内存使用问题

Question

我试图每隔1小时运行一次ML训练脚本，但每小时后内存使用量增加大约20%，3-4小时后，内存使用率达到90%，然后该脚本抛出内存错误。我想知道为什么当火车功能完成时内存不被释放。

虽然如果手动运行train函数(不使用任何类型的线程调度程序并一次或三次调用列车函数)，则不会显示这种行为。

每隔一段时间后训练模型的任何建议。

这是密码。

import pickle
import pandas as pd
from pymongo import MongoClient
import datetime
from apscheduler.schedulers.blocking import BlockingScheduler

def train():

    client = MongoClient(databaseURI)
    db = client['mydb']

    movie_data = []
    for index, obj in enumerate(db.movies.find({})):
        movie_obj = {}

        movie_obj['_id'] = obj['_id']
        movie_obj['title'] = obj['title']
        movie_obj['rating'] = obj['rating']
        movie_data.append(movie_obj)


    user_data = []
    for index, obj in enumerate(db.users.find({})):
        user_obj = {}

        user_obj['_id'] = obj['_id']
        user_obj['username'] = obj['username']
        user_obj['movie_id'] = obj['movie_id']
        user_obj['rating'] = obj['rating']
        user_data.append(user_obj)


    movie_data_df = pd.DataFrame(movie_data)
    user_data_df = pd.DataFrame(user_data)

    # some ML training ALGO
    trainedModel = algo.train(user_data_df, movie_data_df)

    trained.to_pickle('files/trained.pkl')


scheduler = BlockingScheduler()
scheduler.add_job(train, 'interval', hours=1, next_run_time=datetime.datetime.now())
scheduler.start()

Answer 1

作业商店存放预定的作业。默认作业存储只是将作业保存在内存中，而其他作业则存储在各种类型的数据库中。作业的数据在保存到持久作业存储区时被序列化，并在从它加载回时被反序列化。作业存储区(，而不是默认的)不将作业数据保存在内存中，而是充当后端保存、加载、更新和搜索作业的中间人。

我建议尝试以下解决方案之一：

1. 将jobstore从默认的(即内存)更改为某个持久位置(示例)。
2. 或者尝试将参数replace_existing设置为True (默认情况下是False)。 Scheduler.add_job(列车，间隔，hours=1，next_run_time=datetime.datetime.now()，replace_existing=True)

边注：

我认为还可能有另一种丑陋的方法来修复它(我还没有尝试过！)，这样您就可以添加一个Listener来监听崩溃并重新启动整个进程！(如果你能试一试，用一种更轻声的方式来修正它！)

scheduler = BlockingScheduler()
scheduler.add_job(train, 'interval', hours=1, next_run_time=datetime.datetime.now())

def my_listener(event):
    if event.exception:       
        global scheduler
        scheduler.shutdown()
        gc.collect()
        scheduler = BlockingScheduler()
        scheduler.add_job(train, 'interval', hours=1, next_run_time=datetime.datetime.now())
        scheduler.add_listener(my_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
        scheduler.start()

scheduler.add_listener(my_listener, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
scheduler.start()