기본 콘텐츠로 건너뛰기

[Python] 파이썬 무적 스케줄러 소개 - APScheduler


APScheduler란?


APScheduler는 Advanced Python Scheduler의 줄임말로 스케줄을 예약하여 실행할 수 있도록 돕는 파이썬 라이브러리이다. 특정 작업에 주기성을 스케쥴링하는 Airflow등이 있지만, 직관적이며 기존 어플리케이션에서 사용 가능한 큰 장점이있다. 기존 스케줄러 라이브러리와 가장 큰 차이점이자 장점은 while문과 같은 무한루프를 돌리지 않아도 코드를 비동기적으로 실행시킬 수 있다. 여기서 비 동기적이란 코드의 실행 순서에 맞춰 진행하는 것이 아닌 동시 다발적으로 실행하는 것을 말한다. 예를 들면, p2p 사이트에서 다운 받을때 하나씩 받는 것을 동기적, 다수를 한번에 받는것을 비동기적이라 할 수 있다.
본 포스트에서는 본인의 경험을 바탕으로 주요 내용만을 이야기하겠다. 자세한 내용은 apscheduler 공식 user guide에서 확인 가능하다.

✔ 스케줄러 종류

  • BlockingScheduler : 하나의 프로세스에서만 동작하며 스케쥴링 된 작업이 실행될때 다른 작업은 일시중단된다.
  • BackgroundScheduler : 백그라운드에서 동작하며 스케쥴링 된 작업이 실행될때 다른 작업도 할 수 있다.

✔ 실행방식

  • Cron : 주기의 시간(time)에 실행. Date 및 Time 지정이 가능하다
  • Interval : 일정 주기로 실행
  • Date : 특정 날짜로 실행(사실상 Cron과 동일 함)
APScheduler 라이브러리 설치는 아래와 같이 1번줄 명령어로 수행하며, 설치 실패 시 2번줄과 같이 강제로 설치 가능하다.
1
2
pip install APScheduler
pip install --user APScheduler


Blocking Scheduler

Blocking Scheduler의 사용 방법은 다음과 같다. 주의할 점은 sched.start()다음은 실행되지 않는다. Scheduler.job 함수의 자세한 내용은 참고문헌에서 확인 가능하다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import time
from apscheduler.schedulers.blocking import BlockingScheduler
 
# 스케줄링할 작업을 함수로 정의
def job1():
    return print(f'job1 : {time.strftime("%H:%M:%S")}')
 
def job2():
    return print(f'job2 : {time.strftime("%H:%M:%S")}')
 
# 스케쥴러
sched = BlockingScheduler(timezone='Asia/Seoul')
 
# 매 5초 주기로 실행 - interval
sched.add_job(job1, 'interval', seconds=5, id='test_1')
 
# 매분에 실행 - cron
sched.add_job(job2, 'cron', second='0', id="test_2")
 
print('sched before start')
sched.start()
print('sched after start') # Blocking 이기 때문에 실행 안됨


Background Scheduler

필자는 여러 공공 데이터의 API를 주기적으로 호출하기 위해 Background Scheduler를 유용하게 사용하였다. 자세한 Background Scheduler의 사용 방법은 다음과 같으며 Blocking Scheduler와 비교하기 위해 기존 코드를 수정하였다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import time
# from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.schedulers.background import BackgroundScheduler
 
# 스케줄링할 작업을 함수로 정의
def job1():
    return print(f'job1 : {time.strftime("%H:%M:%S")}')
 
def job2():
    return print(f'job2 : {time.strftime("%H:%M:%S")}')
 
# 스케쥴러
# sched = BlockingScheduler()
sched = BackgroundScheduler(timezone='Asia/Seoul')
 
# 매 10초 주기로 실행 - interval
sched.add_job(job1, 'interval', seconds=10, id='test_1')
 
# 매분에 실행 - cron
sched.add_job(job2, 'cron', second='0', id="test_2")
 
print('sched before start')
sched.start()
print('sched after start') # 살행 됨
 
while True:
    time.sleep(1)

예상치 못한 자원 부족 에러

(1) misfire_grace_time

기존 Scheduler 라이브러리로 실행했을 경우 에러 발생시 종료되는 문제로 골머리를 쓰다 신처럼 강림한 APScheduler를 만난 후 내 삶은 편해졌다.

CPU 사용량이 많거나 메모리 부족으로 간혹 사용 가능한 Thread가 부족하여 에러가 발생하지만 시스템 기본 값에 의해 다시 실행된다. (기존 Scheduler 라이브러리와는 가장 큰 차이점인 것 같다)
Thread 부족 에러는 보통 add_job내 misfire_grace_time* 설정을 안했을 경우 종종 다음과 같은 에러가 발생하며, Job내 파라미터 설정시 사용자 정의 시간(또는 지연시간) 이후 Job이 정상 실행될 수 있도록 한다.
(*misfire_grace_time: 누락된 작업 실행 가능한 시간, 기본 값: undefined object)

Run time of job "job1 (trigger: interval[0:00:10], next run at: 2024-02-12 00:45:33 KST)" was missed by 0:00:01.233551

Thread의 자세한 개념은 참고문헌에서 확인 가능하다.

(2) max_instances

또 열심히 돌리보면 아래와 같은 경고도 발견할 수 있다.

WARNING:apscheduler.scheduler:Execution of job "job1 (trigger: interval[0:00:10], next run at: 2024-02-12 11:50:42 UTC)" skipped: maximum number of running instances reached (1)

인스턴스 오류가 발생하여 해당 Job을 수행할 수 없어서 다음 시간에 다시 실행 한다는 건데, 이러한 오류도 보기 싫으면 간단한 매개변수 추가로 문제를 해결할 수 있다. 
1
scheduler.add_job(job, 'interval', seconds=10, max_instances=2)

Labels:

댓글

댓글 쓰기

이 블로그의 인기 게시물

[AI] RAG란 무엇인가?

자고 일어나면 인공지능 관련 기사와 내용이 참 많습니다. 소위 AI시대 대격변 속에서 무엇을 해야할지 고민이 많습니다. AI의 기본적인 이론도 중요하지만, 최근 트렌드를 잘 알고 잘 써먹을 수 있는 것도 개인의 역량이 아닐까 싶습니다. 즉 AI를 하나의 도구로 생각하고 도구를 잘 써먹으면 좋지 않을까요? 오늘은 RAG(검색-증강-생성)이란 무엇인지 간략하게 끄적여보겠습니다. 3줄 요약 ⓐ 검색(Retrieval)은 요청된 외부 지식 을 가져온다. ⓑ 증강(Augmented)은 질문 것에 요청된 것을 더한다 . ⓒ 생성(Generation)은 사용자의 질문에 더한 것을 받아 텍스트로 생성 한다.   ⅰ RAG(검색-증강-생성)이란 무엇인가? 최근 인공지능 기술의 발전은 다양한 분야에서 혁신을 일으키고 있으며, 그 중에서도 텍스트 기반의 생성형 AI는 특히 주목받고 있습니다. 하지만 이러한 인공지능 기술이 가진 정확성과 신뢰성의 문제는 사용자와 개발자 모두에게 중요한 과제로 남아 있습니다. 이 문제를 해결하기 위해 등장한 기술 중 하나가 바로 RAG(Retrieval-Augmented Generation)입니다. 자세한 기술적 내용은 RAG 용어가 등장한 2020년 아래의 논문을 참고하기 바랍니다. Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks(2020) (본지는 간단한 내용 만을 기술할 목적이기 때문에 IT 기업의 설명 자료를 참고하여 본인이 이해한 내용을 작성 하였습니다. 혹시나 잘못된 정보가 있다면 알려주세요.) RAG(Retrieval-Augmented Generation)는 단어 풀이 그대로 '검색 증강 생성'이란 뜻 입니다. 먼저 Retrieval 이란 의미는 검색이란 뜻보다는 외부 지식 데이터베이스에서 가져오는 것, 어딘가에서 요청된 무엇인가를 가져오는 것 을 이야기합니다. 그리고 Augmented 는 증강이란 뜻으로 원래 것에 뭔가를 더하거나 보태어 더 실해졌다 는 ...
댓글 1개
Read more »

[Machine Learning] 경사 하강법 (Gradient Descent)

모델의 오류는 왜 중요할까? 들어가기 전에 한 줄 요약해보겠습니다. 의심하며 본문의 내용을 읽어주세요. 혹시나 다른 내용이 있다면 덧글 부탁드립니다. 모델의 계수를 구하기 위해 가중치를 바꿔가며 전역(global) 기울기가 0인 곳을 잘 찾아보자 . 들어가기 머신러닝 혹은 딥러닝의 가장 적합한 모델 이라 함은 대부분 모델의 오류(error)를 최소화 하는 것을 의미한다. 즉, 어떤 모델(단변량 함수)의 계수의 최적값을 찾는 것으로 생각 할 수 있다. 최적화 문제를 풀기위해 경사 하강법 이라 부르는 방법을 사용할텐데, 이는 곧 함수의  기울기 를 계산하고 경사의 이동방향의 반대 방향으로 이동하여 극값 (=0) 을 찾는 문제 와 같다. 접선의 오류를 찾는 손실함수는 다음 포스트에서 다루도록 하겠다. (즉 머신러닝/딥러닝 최적 모델은 함수의  기울기와 접선의 오차를 줄이는  것이다) 본 포스트에서는 경사 하강법의 기본 개념에 한하여 설명하도록 하겠다. 우리가 생각하는 1차원 혹은 2차원에서의 기울기 최소값이 아닌 아래 그림과 같이 3차원 이상의 다차원에서는 함수에 하나의 전역 최소값(global minimum) 이 존재하는 경우 꽤 쓸만하지만, 함수에 지역 최소값(local minimum) 여러 개 있는 경우에는 시작점을 어디에 잡냐에 따라 잘못된 곳으로 빠질 수 있다. (*딥러닝의 경우 때에 따라 다차원 함수의 최소값을 찾아야 한다. 이는 인간의 머리로는 한계가 있다.) Python 코드를 활용해 간단한 그라디언트 함수를 만들어 보자. f가 단변수 함수인 경우, 점 x 에서의 미분값은 x가 아주 조금 변했을 때 f(x)의 변화량 을 의미한다. x의 변화량을 식에서는 h로 표기한다. 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 from  typing  import  Callable def  difference_quoitemt(f: Callab...
댓글 쓰기
Read more »

[Machine Learning] 데이터 증강 (Data Augmentation) 이란?

데이터 증강 (Data Augmentation)이란? 본격적으로 들어가기전 내용을 간략하게 요약하자면 다음과 같다. 요약을 바탕으로 본문에서는 좀더 깊이있게 설명하도록 하겠다. 이미지 데이터의 증강은 회전, 크기 조절, 색생 변화 등 다양한 변환을 통해 학습데이터의 다양성을 인위적으로 늘려 모델의 성능을 형상 시키는 방법 시계열 데이터의 증강은 시간 축 변환, 크기 조절, 시간 이동 등 다양한 기법을 통해 원본 데이터를 변형하여 학습 데이터의 다양성을 높이고 모델의 일반화를 향상 시키는 기법 CNN, R-CNN, YOLO 등 모델의 성능을 높이고 오버피팅을 극복할 수 있는 가장 좋은 방법은 다양한 유형의 학습 이미지 불균형 극복을 위한 데이터 양을 늘리는 것입니다. 하지만 사진, 동영상과 같은 이미지 데이터의 경우 학습 데이터 량을 늘리는 것은 쉽지가 않다. 이미지 데이터의 양을 늘리긴 위해서는 결국 클라스(라벨링)의 수도 증가하며, 이는 곧 엄청난 노가다가 필요하기 때문이다. 데이터 증강은 학습 이미지의 개수를 늘리는 것이 아니고 학습 시(epoch) 마다 개별 원본 이미지를 변형해서 학습하는 것 이다. 아래 첨부의 그림과 같이 학습 이미지를 변형할 수 있다. (출처:  https://blog.insightdatascience.com/automl-for-data-augmentation-e87cf692c366) 데이터 증강을 우리 일상생활로 쉽게 설면하자면 우리가 핸드폰 카메라를 통해 사진을 촬영하고 사진의 밝기, 크기, 좌우반전 등을 행하는 것을 이미지 데이터의 증강으로 이해하면 쉽다. tensorflow 에서 간단한 예제를 활용하여 이미지 데이터의 증강을 설명하고 있다. 본 블로그에서는 몇 가지 대표적인 데이터 증강 종류를 간단히 설명하고 마무리 하겠다. 기회가 된다면 텍스트 및 시계열 데이터의 증강 방법도 간략히 소개 후 마무리 하겠다. 이미지 데이터의 증강 (출처:  https://www.invivoo.com/ ) 이미지에 아래와 같이...
댓글 쓰기
Read more »