[Python] APScheduler 를 이용한 파이썬 스케줄링 스크립트

ML 파이프라인 자동화를 하면서 특정 시간에 주기적으로 돌려야하는 스케쥴러가 필요했다.

파이썬 코드로 실행을 해야해서 어떤 스케쥴러를 사용할까 찾아보다 APScheduler에 대해 찾아보게 됐다.

 

APScheduler (Advanced Python Scheduler)

• Python code 를 주기적으로 수행할 수 있게 도와주는 Python Library

Schedule Type

• cron : Cron 표현식으로 Python code 를 수행
• interval : 일정 주기로 Python code 를 수행
• date : 특정 날짜에 Python code 를 수행

Scheduler 종류

대표적으로 

• BlockingScheduler : 단일 Job 수행시
• BackgroundScheduler : 다수 Job 수행시

APSchedule 설치

• pip install apscheduler

 

예시 코드

해당 코드는 BlockingScheduler를 사용했다.

add_job 함수 내부에 시간에 대한 파라미터가 들어간다.

테스트 도중 CPU 사용량이 많거나 메모리가 부족해서 정해진 시간에 작업이 진행될 수 없는 경우가 있었는데 misfire_grace_time을 60으로 설정해줌으로써 60초정도 지연시간을 두어 정상 처리될 수 있도록 했다. 

import argparse
from apscheduler.jobstores.base import JobLookupError
from apscheduler.schedulers.background import BlockingScheduler
import time
from datetime import datetime, timedelta

class Scheduler:
    def __init__(self):
        self.sched = BlockingScheduler()
        
    def test(self):
        print(datetime.now())

    def scheduler(self, args):
        job_type = args.type
        job_day = args.day
        job_hour = args.hour
        job_min = args.min
        print("{type} Scheduler Start".format(type=job_type))
        print(args)

        if job_type == 'cron':
            self.sched.add_job(self.test, job_type, day_of_week=job_day,
                               hour=job_hour, minute=job_min, misfire_grace_time=60)
            self.sched.start()
        elif job_type == 'interval':
            self.sched.add_job(self.test, job_type, day_of_week=job_day,
                               hour=job_hour, minute=job_min, misfire_grace_time=60)
            self.sched.start()
            
if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    # python scheduler.py --type cron --day Mon-Sun --hour 18 --min 30
    
    parser.add_argument("--type", default="cron", type=str,
                        choices=['cron', 'interval'], help="The type of the scheduler")
    parser.add_argument("--day", default="Mon-Sun", type=str,
                        help="The day of the scheulder")
    parser.add_argument("--hour", default="00", type=str,
                        help="The day of the scheulder")
    parser.add_argument("--min", default="0", type=str,
                        help="The minute of the scheulder")
    
    args = parser.parse_args()

    try:
        scheduler = Scheduler()
        scheduler.scheduler(args)
    except Exception as msg:
        print(msg)