대화형 파이프라인

대화형 파이프라인

머신러닝 파이프라인 전체를 설계하고 구현하는 일은 다소 어려운 일 중 하나이다. 한 번에 에러 없이 잘 동작하는 파이프라인을 구축하는 것은 프로젝트가 거대해질 경우 거의 불가능에 가깝고, 컴포넌트를 디버깅하는 일조차 일체형 파이프라인에서는 쉽지 않다.

 

이러한 문제를 해결하기 위해 TFX는 대화형 파이프라인 기능을 도입하였다. 대화형 파이프라인은 단계별 구현이 가능하며 구성 요소의 아티팩트를 즉시 검토할 수 있다는 장점이 있다. 각 단계의 구현과 디버깅이 완료된 후에는 아파치 에어플로에서 실행할 수 있도록 대화형 파이프라인을 실제 프로덕트 레벨의 파이프라인으로 변환하는 과정을 거치면 전체 파이프라인의 성공적인 구축이 완료된다고 할 수 있다.

---

실습

모든 대화형 파이프라인은 주피터 노트북이나 코랩 등의 콘텍스트에서 프로그래밍되며, 오케스트레이션 툴과 달리 사용자가 명령을 입력하고 실행하는 방식이다.

 

import tfx
import tensorflow as tf
from tfx.orchestration.experimental.interactive.interactive_context import InteractiveContext

💡 pip install tfx==1.2.0을 실행했음에도 위 코드가 정상적으로 동작하지 않는다면 런타임 다시 시작을 해보자.

 

이제 콘텍스트 객체를 생성할 수 있다. 콘텍스트 객체는 컴포넌트 실행을 처리하고 컴포넌트의 아티팩트를 표시한다. 이때 InteractiveContext는 간단한 메모리 내 ML 메타데이터스토어도 설정한다.

 

context = InteractiveContext()

 

파이프라인 컴포넌트를 설정한 후 다음 예제와 같이 콘텍스트 객체의 run 함수로 각 컴포넌트 객체를 실행할 수 있다.

 

from tfx.components import StatisticsGen

statistic_gen = StatisticsGen(
    examples=example_gen.outputs['examples']
) # 당연스럽게도, 아직 example_gen을 define하지 않아 실행은 불가하다

context.run(statistic_gen)

 

컴포넌트 객체는 StatisticGen의 매개변수로 이전 컴포넌트(위 예시에서는 데이터 수집 컴포넌트인 ExampleGen)의 출력을 수신한다. 컴포넌트의 태스크를 실행한 후, 컴포넌트는 출력 아티팩트의 메타데이터를 메타데이터스토어에 자동으로 기록한다. 

 

대화형 파이프라인의 장점 중 하나는 중간 출력을 바로 확인할 수 있는 점이라고 언급하였다. 즉, 일부 컴포넌트의 출력의 경우, 대화형 셀을 이용해 출력이 가능하다. 예를 들어, 다음처럼 StatisticGen 컴포넌트를 사용하여 데이터셋의 feature를 검사할 수 있다.

 

context.show(statistic_gen.outputs['statistics'])

 

완성된 예제는 다음과 같다.

 

import os
from pathlib import Path

import tfx
from tfx.components import StatisticsGen
import tensorflow as tf
from tfx.orchestration.experimental.interactive.interactive_context import InteractiveContext

context = InteractiveContext()

DATA_PATH = 'https://raw.githubusercontent.com/tensorflow/tfx/master/tfx/examples/chicago_taxi_pipeline/data/simple/data.csv'

dir_path = Path().parent.absolute()
data_dir = os.path.join(dir_path, '..', '..', 'data', 'taxi')

Path(data_dir).mkdir(parents=True, exist_ok=True)

# download raw data
filepath = tf.keras.utils.get_file(
    os.path.join(data_dir, 'data.csv'),
    DATA_PATH
)

# ExampleGen
example_gen = tfx.components.CsvExampleGen(input_base=data_dir)
context.run(example_gen)

# StatisticGen
statistic_gen = StatisticsGen(examples=example_gen.outputs['examples'])
context.run(statistic_gen)

# inspect data set's features
context.show(statistic_gen.outputs['statistics'])

# inspect component's output artifact with programming method
statistic_gen = StatisticsGen(examples=example_gen.outputs['examples'])
context.run(statistic_gen)

for artifact in statistic_gen.outputs['statistics'].get():
  print(artifact.uri)

대화형 파이프라인을 이용해 train set의 요약을 확인한 모습