파이썬으로 구현하는 스트림리더

파이썬에서 파일의 내용을 읽어와서 처리할 때 가장 기본적인 방법은 opne() 함수를 이용해서 파일 객체를 만들고, read() 메소드를 이용해서 파일의 전체 내용을 한 번에 읽어오는 것이다. 그런데 많은 경우에 실제로 다루는 파일은 텍스트포맷인 경우가 많다. 텍스트 포맷을 다룰 때에는 다음과 같은 몇 가지 전략이 존재한다.

  • read()를 이용해서 파일의 전체 내용을 읽어와 하나의 문자열로 사용한다.
  • readlines()를 이용하면 파일의 전체 내용을 읽어와서 라인 단위로 잘라 문자열의 리스트로 만들 수 있다.
  • readline()을 이용해서 파일의 내용을 한 줄씩 읽어와서 처리한다.

실제로는 텍스트 파일을 한줄 단위로 읽어와서 파싱해서 사용하는 경우가 많기 때문에 위 방법 중에서는 readline()이 가장 많이 쓰이며, 아예 텍스트 파일을 연 파일 객체는 문자열의 제너레이터처럼 동작하기 때문에 for ... in 구문으로 한줄씩 데이터를 읽어서 사용한다. 이 방식의 가장 좋은 점은 파일을 한 줄 단위로만 읽어와서 처리하기 때문에 불필요한 메모리 낭비를 하지 않는다는 것이다. 심지어 수 기가짜리 텍스트 파일이 있더라도, 한 줄씩 처리하는 경우에는 메모리에 한줄 만큼의 분량을 읽어와서 처리하기 때문에 프로그램이 메모리 부족으로 죽을 일이 없다.

그런데 어떤 경우에는 텍스트로 구성된 대용량 파일이 있고, 이 파일 내의 데이터가 컴마나 스페이스 같은 문자만 구분되어 있고 개행문자가 포함되지 않았다면 어떻게 처리하면 좋을까? 극단적인 예이기는 하지만 몇 기가 짜리 텍스트 파일이 있고, 여기에는 컴마로 구분된 숫자값들이 있다고 하자. 이를 읽어서 split()으로 나눠서 써야겠지만, split()이라는 동작 자체가 일단 잘라낼 문자열을 메모리에 올린 다음, 한꺼번에 잘라서 리스트로 만드는 방식으로 처리되기 때문에 실제 문자열의 크기의 두 배 이상의 메모리를 요구하게 된다.

이 문제를 해결하기 위해서 조금 느긋하게 파일의 내용을 읽어들이는 방법을 구현해보자. 기본적인 아이디어는 다음과 같다.

  1. 파일의 내용을 일부 읽어올 버퍼를 준비한다.
  2. 미리 정해둔 사이즈만큼 파일의 내용을 읽어와서 버퍼에 담는다.
  3. 버퍼의 처음부터 구분자가 있는 곳까지를 탐색한 후, 해당 영역을 잘라내어 사용한다.
  4. 구분자를 버린다.
  5. 3의 과정을 반복한다. 만약 구분자가 발견되지 않으면, 파일로부터 다시 정해진 사이즈만큼의 데이터를 읽어와서 버퍼에 덧붙인다.
  6. 파일에서 더 이상 읽을 내용이 없다면 버퍼에 남아있는 데이터가 마지막 조각이 된다.
  7. 파일을 닫는다.

이처럼 파일로부터 특정한 구분자까지의 데이터를 읽어서 순차적으로 잘라내어 리턴해주는 제너레이터를 만들 수 있을 것이다.

def read_strem(filename, sep=","): 
    ## filename : 읽을 파일
    ## sep : 구분자
    buffer = bytearray()
    chunk_size = 1024 # 한 번에 1024바이트씩 읽어온다. 
    token = sep.encode()

    with open(filename, 'rb') as f: ## 파일을 이진 파일로 읽어온다. 
      while True:
        data = f.read(chunk_size)
        if not data: ## 파일에서 더이상 읽을 내용이 없으면 루프 종료
          break
        buffer.extend(data)
        ## 버퍼내에서 구분자까지 자르는 작업을 반복
        while True:
          i = buffer.find(token)
          if i < 0 :  ## 구분자가 발견되지 않으면 한 번 더 읽어온다.
            break
          found = buffer[:i].decode()
          yield found
          ## 구분자앞까지 자른 내용을 내놓은 후에는 버퍼의 앞쪽을 정리한다.
          buffer[:i+len(token)] = []
       ## 파일에서 읽는 내용을 모두 처리했다. 버퍼에 남은 내용이 있으면 내놓는다.
       if buffer:
         yield buffer.decode()

만약 아주 커다란 텍스트파일로부터 컴마로 나누어진 데이터를 읽어와서 처리하는 동작을 수행한다고 해보자. 심지어 어떤 경우에는 데이터 전체가 필요한 것도 아니고 앞 부분에서 1000개 정도만 필요할 수도 있다. 위에서 만든 함수는 제너레이터 함수이기 때문에 for 문에서 사용할 수 있고, 만약 앞 1000개까지만 사용하려면 다음과 같이 쓸 수 있다.

## 파일이름이 'verybigfile.txt'라 할 때,
g = read_stream('verybigfile.txt')
for (i, w) in enumerate(g):
    if not i < 1000:
       break
    print(w)

약간 다른 접근방법도 있다. 파이썬보다는 Objective-C나 Swift에 어울릴 것 같은 방법인데, “필요할 때마다 꺼내 쓰는” 제너레이터가 아니라 튀어나오는 데이터를 처리할 핸들러를 넘겨주고 함수 내에서 다 처리해버리는 방법이다. 핸들러의 디자인은 대략 다음과 같은 식으로 처리하면 되겠다.

  • 처리할 문자열 데이터와, 몇 번째 데이터인지를 나타내는 정수 값을 인자로 받고
  • 문자열을 처리한 후, 더 처리할 것인지 그렇지 않은지를 리턴한다. None이나 False로 평가되는 값을 리턴하면 계속하고, 그렇지 않으면 더 이상 실행하지 않도록 할 것이다.

예를 들어 다음과 같은 식으로 핸들러를 만들 수 있는데,

def process_word(word, idx):
  if idx < 10:
     print(word)
     return
  return True

위 핸들러는 10번째까지는 출력하고, 그 이후로는 더 이상 실행하지 않겠다는 의미가 된다. 이런 핸들러를 받아서 처리해주는 형태로 위 제너레이터 함수를 다시 쓰면 다음과 같은 모양이 될 것이다.

def split_file(filename, handler, sep=","):
  buffer = bytearray()
  chunk_size = 1024
  token = sep.encode()
  idx = 0
  with open(filename, 'rb') as f:
    while True:
      data = f.read(chunk_size)
      if not data:
        break
      buffer.extend(data)
      while True:
        i = buffer.find(token)
        if i < 0:
          break
        found = buffer[:i].decode()
        r = handler(found, idx)
        if r:
          return
        buffer[:i+len(token)], idx = [], idx+1
    if buffer:
      handler(buffer.decode(), idx)

## 10개 데이터만 출력하고 끝낸다.
split_file('varybidfile.txt', process_word)

이 알고리듬은 제법 간단하면서도 대용량 데이터 파일을 안전하게 처리할 수 있도록 해주는 제법 괜찮은  방법이다. 간단하기 때문에 Swift 등의 다른 언어로도 충분히 컨버팅할 수 있다.