Intro
Python에서 Lambda, Iterator, Yield 등 여러 방법으로 프로그래밍을 할 수 있도록 여러 기능들을 제공해줍니다. Lambda와 Iterator 등의 기능은 일반적인 경우에 함수 대신, 혹은 반복문 사용 시 자주 사용하기에 익숙했습니다. 그러나 최근 Yield에 대한 개념이 머릿속에서 흐릿하게 지워지고 있어, 다시 한 번 이 기능을 활용하여 개발해보고자 하는 욕심이 생겼습니다. 그렇게 기회를 살피다 최근 성능개선 과정에서 Yield를 사용할 기회가 생겼고, 이를 어떻게 활용했는지 간단하게 공유하고자 합니다.
Yield?
먼저 간단하게 python에서 제공하는 yield 기능에 대해 알아보도록 합시다. 블로그 포스트나 점프투파이썬 개념을 학습할 때 다음과 같은 예제를 살펴보셨을 거라 생각합니다. 아래의 예제에서는 yield를 이용하여 1,2,3 값을 순차적으로 반환하는 로직입니다.
Basic yield
def yield_test():
yield 1
yield 2
yield 3
gen = yield_test()
print(type(gen)) # <class 'generator'>
print(next(gen)) # 1
print(next(gen)) # 2
print(next(gen)) # 3
print(next(gen)) # StopIteration...
Python
복사
이 과정에서 yield_test() 함수는 generator 타입이며, next 함수로 iterable한 객체를 호출하여 값을 가져올 수 있습니다. 일반적인 함수의 경우 값을 return 받을 때 함수를 종료하고 값을 반환하는 형태를 취하지만, yield를 사용하게 될 경우 generator 형태로 값을 순차적으로 반환해주는 형태로 선언할 수 있습니다.
Generator
Generator란 iterator를 생성해주는 함수이며, 함수 안에서 yield 키워드를 사용합니다. 특징은 다음과 같습니다.
•
iterable한 순서가 지정됨
•
순서의 다음 값은 필요에 따라 계산됨
•
함수의 내부 로컬 변수를 통해 내부상태가 유지됨
•
무한한 순서가 있는 객체를 모델링할 수 있음
◦
끝없는 데이터 스트림
◦
Random Bit Generator
Examples
다른 예제를 가져와 조금 더 세밀하게 살펴보도록 합시다. 만약 ‘숫자 0 ~ 122까지의 배열에서 짝수만 필터링하여 출력하라’ 라는 문제를 풀이할 때 일반적인 방법과 yield를 활용하는 경우를 구분하여 작성해보았습니다.
Basic solve
일반적으로는 숫자를 가지고 있는 배열을 함수에서 전달받아 필터링을 마친 후 그 결과값을 반환하여 main에서 사용하는 형태를 가지고 있을 것입니다. 하지만 아래의 예제처럼 사용할 경우 get_even_number 함수가 끝날 때까지 main 동작이 멈춰 있는 형태가 됩니다.
def get_even_number(data):
result = []
for d in data:
if not d % 2:
result.append(d)
return result
if __name__ == "__main__":
TEST_DATA = []
for i in range(123):
TEST_DATA.append(i)
nums = get_even_number(TEST_DATA)
for n in nums:
print(n, end=", ")
# 2, 4, 6, 8 .... 120, 122
Python
복사
위 코드를 풀어서 Flow-Chart로 그려보면 아래와 같이 표현해볼 수 있을 것 같습니다.
yield solve
만약 yield 를 사용할 경우 다음과 같이 코드가 약간 더 간결해지고, yield만의 장점을 취할 수 있습니다. 아래의 코드에서는 result라는 추가 배열이 없이 그때그때 적절한 값을 반환해주는 형태로 개발되어 있습니다. 이 경우 장점을 취할 수 있습니다.
•
get_even_number 내에서 추가 메모리(result)를 사용하지 않아도 됨
•
main 함수에서도 추가 메모리(nums)를 사용하지 않아도 됨
def get_even_number(data):
for d in data:
if not d % 2:
yield d
if __name__ == "__main__":
TEST_DATA = []
for i in range(123):
TEST_DATA.append(i)
for n in get_even_number(TEST_DATA):
print(n, end=", ")
# 2, 4, 6, 8 .... 120, 122
Python
복사
위의 코드를 FLow-Chart로 그려보면 다음과 같이 표현해볼 수 있을 것 같습니다.
Return vs Yield
두 개의 Flow-Chart로 비교해보면 크게 차이가 없어보이지만, 가장 큰 차이점이라고 한다면 값을 모두 처리한 후 return하거나 값을 그때 그때 조건에 맞춰 yield하는 부분일 것입니다.
Problem solves
이러한 원리를 이용하여 문제를 해결하고자 하는 문제는 다음과 같습니다.
•
특정 개수의 메시지를 예약하여 Database에 저장되어 있음
•
예약 시간에 메시지를 보낼 때 특정 개수(slice_size)만큼 나누어서 보냄
•
특정 개수 만큼 발송하고 일정 시간(slice_term) 동안 sleep 후 발송을 반복함
As-Is
기존의 소스코드는 다음과 같습니다. Interface 내에서 chunker 부분에서 일정 크기로 보내고자 하는 instance를 나누고 sends 로직을 불러와 발송하는 코드입니다. 여기서 또 하나의 미션으로, 기존의 소스코드는 최대한 수정하지 않고 기능을 개선이 라는 제약사항을 두었습니다. 이러한 제약사항을 두고 기능을 개선하고자 했을 때 떠오르는 가장 좋은 방법은 Yield 만한 게 없다였습니다.
from typing import OrderedDict, Generator, List
class Interface:
def __init__(self, settings):
self._settings = settings
self.CHUNK_SIZE = 1000
def chunker(instances: OrderedDict) -> Generator:
size = self.CHUNK_SIZE
return (instances[pos:pos+size] for __ in range(0, len(instances), size))
def sends(chunked_instances: List):
for __ in chunked_instances:
# send logic....
print(f"Send....")
if __name__ == "__main__":
interface = Interface(SOMTHING_SETTINGS)
for chunked in interface.chunker(instances):
# Send chunked ...
sends(chunked)
"""
Output:
>> Send....
>> Send....
>> Send....
>> Send....
>> Send....
"""
Python
복사
To-Be
최대한 기존 기능에서 수정 없이 개선하기 위해서는 다음과 같이 개선해보았습니다. 개선된 기능을 비교하기 위해 chunker와 changed_chunker로 구분해보았습니다.
import time
from typing import OrderedDict, Generator, List
class Interface:
def __init__(self, settings):
self._settings = settings
self.CHUNK_SIZE = 1000
def chunker(instances: OrderedDict) -> Generator:
size = self.CHUNK_SIZE
return (instances[pos:pos+size] for __ in range(0, len(instances), size))
def changed_chunker(instances: OrderedDict) -> Generator:
size = self.CHUNK_SIZE
for index, chunked in enumerate(instances[pos:pos+size] for __ in range(0, len(instances), size)):
yield chunked
if (index+1) % int(self._settings.SLICE_SIZE/size) == 0:
if not int(len(seq)/size) == index:
# If loop is done. no sleep.
time.sleep(self._settings.SLICE_TERM)
def sends(chunked_instances: List):
for __ in chunked_instances:
# send logic....
print(f"Send....")
if __name__ == "__main__":
"""
ADD {"SLICE_SIZE": int, "SLICE_TERM": int} into SOMETHING_SETTINGS
"""
start = time.time()
interface = Interface(SOMTHING_SETTINGS)
for chunked in interface.changed_chunker(instances):
# Send chunked ...
sends(chunked)
print(f"sleep - {time.time() - start:.3f}")
"""
Output:
>> Send....
>> Send....
>> sleep - 5.003s
>> Send....
>> Send....
>> sleep - 10.007s
>> Send....
"""
Python
복사
Conclusion
힘이 빠지는 말이지만, 위와 같은 방법으로 문제를 해결해보려고 했으나 더 좋은 방법이 떠올라 수정하였습니다… 이번 글에서는 기존의 코드를 최대한 수정하지 않고 요구사항을 수정해보는 방법을 고민해보고 적절한 방법을 찾아보는 과정, 그리고 Yield의 기본 개념을 함께 풀어서 작성해보았습니다.
이번 경험으로 요구사항을 구현하기 위해 적절한 방법을 적용하기 위해서는 코드의 흐름을 파악하는 것이 좋다는 점을 배울 수 있었습니다. 이번 포스팅 방법을 실제 코드에 적용하지 않은 이유에 대해서는 다음 글에서 다뤄보고자 합니다.