Stream

스트림(Stream)이란?

• 컴퓨터 처리 환경에서 스트림(stream)은 시간이 지남에 따라 사용할 수 있게 되는 일련의 데이터 요소를 가리키는 수많은 방식에서 쓰임
  • C 프로그래밍 언어에 기반을 둔 유닉스 관련 시스템에서 스트림은 개별 바이트나 문자열인 데이터의 원천
    • 스트림들은 파일을 읽거나 쓸 때, 네트워크 소켓을 거쳐 통신할 때 쓰이는 추상적인 개념
    • 표준 스트림들은 모든 프로그램에 이용할 수 있는 세 개의 스트림을 말함
  • 파이프라인은 장치에 삽입된 제한이 없는 정보 뿐 아니라 스트림으로 이해할 수 있음
  • 스킴 프로그래밍 언어 등에서 스트림은 느긋하게 계산하거나 지연 처리된 일련의 데이터 요소를 말함
    • 스트림은 리스트와 유사하게 사용되지만 나중에 이 요소들은 필요할 때에만 계산함
    • 그러므로 스트림은 무한 수열과 급수를 대표할 수 있음
  • 스트림 프로세싱: 병렬 컴퓨팅에서, 특히 그래픽 처리에서 스트림이라는 용어는 소프트웨어뿐 아니라 하드웨어에도 적용

Tip! 추가 내용

• 일반적으로 데이터, 패킷, 비트 등의 일련의 연속성을 갖는 흐름을 의미
  • 음성, 영상, 데이터 등의 작은 조각들이 하나의 줄기를 이루며 전송되는 데이터 열(列)
  • 호스트 상호간 또는 동일 호스트 내 프로세스 상호간 통신에서 큐에 의한 메세지 전달 방식을 이용한 가상 연결 통로를 의미하기도 함

Python 파일 작업

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

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

텍스트 포맷을 다루는 전략

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

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

Tip! 파싱(Parsing)이란?

• 문서나 HTML 등 어떤 큰 자료에서 내가 원하는 정보만 가공하고 추출해서 원할 때 불러올 수 있게 하는 것
• 특정 패턴이나 순서로 추출해서 정보로 가공할 수 있고 XML, DOM, SAX, JSON 등과 같은 파싱 기법이 있음
• 파싱을 수행하는 프로그램은 파서(Parser)라고 함

• 그런데 어떤 경우에는 텍스트로 구성된 대용량 파일이 있고, 이 파일 내의 데이터가 컴마나 스페이스 같은 문자만 구분되어 있고 개행문자가 포함되지 않았다면 어떻게 처리하면 좋을까?
  • 극단적인 예이기는 하지만 몇 기가 짜리 텍스트 파일이 있고, 여기에는 컴마로 구분된 숫자값들이 있다고 했을 때
  • 이를 읽어서 split()으로 나눠서 써야겠지만, split()이라는 동작 자체가 일단 잘라낼 문자열을 메모리에 올린 다음, 한꺼번에 잘라서 리스트로 만드는 방식으로 처리되기 때문에 실제 문자열의 크기의 두 배 이상의 메모리를 요구하게 됨
  • 문제 해결 방법
    • ① 파일 내용을 일부 읽어올 버퍼를 준비
    • ② 미리 정해둔 사이즈만큼 파일의 내용을 읽어와서 버퍼에 담음
    • ③ 버퍼의 처음부터 구분자가 있는 곳까지를 탐색한 후, 해당 영역을 잘라내어 사용
    • ④ 구분자를 버림
    • ⑤ ③의 과정을 반복
      • 만약 구분자가 발견되지 않으면, 파일로부터 다시 정해진 사이즈만큼의 데이터를 읽어와서 버퍼에 덧붙임
    • ⑥ 파일에서 더 이상 읽을 내용이 없다면 버퍼에 남아있는 데이터가 마지막 조각이 됨
    • ⑦ 파일을 닫음
  • 이처럼 파일로부터 특정한 구분자까지의 데이터를 읽어서 순차적으로 잘라내어 리턴해주는 제너레이터를 만들 수 있음

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)