Advanced Python Scheduler

Advanced Python Scheduler (APScheduler)

Python으로 이뤄진 전처리나 ETL 을 주기적으로 실행해야하는 경우, dynamic task scheduler가 팔요하다. 이에 대한 선택지로는 Airflow나 dask, zenkins 등이 있으나, 매우 간단하게 single machine에서의 job scheduling으로는 좀 오버스팩이 아닐 수 없다. 이러한 경우에 APS를 사용하면 편리하다.

설치방법

pip install apscheduler

API문서

https://apscheduler.readthedocs.io/en/stable/modules/schedulers/base.html#module-apscheduler.schedulers.base

간단한 예제

from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler
from simpduler.scheduler.execute import run_all

def get_the_first_schedule():
    now = datetime.now()
    date_time = now.strftime("%Y-%m-%d")
    date_time = date_time + " 03:00:00"
    print("Next schedule:", date_time)
    return date_time


def build_scheduler():
    scheduler = BlockingScheduler()
    scheduler.add_executor('processpool')
    scheduler.add_job(run_all, 'interval', days=1, start_date=get_the_first_schedule())
    
    try:
        scheduler.start()
    except (KeyboardInterrupt, SystemExit):
        pass