Building ML Pipelines 따라잡기 (2) - TFX(Tensorflow Extended)

(포스팅의 모든 내용은 Hannes Hapke, Catherine Nelson 의 저서인 Building Machine Learning Pipelines(O’Reilly, 2020)를 직접 읽고 작성한 후기 및 정리글입니다.)

TFX(Tensorflow Extended)

TFX(Tensorflow Extended) 라이브러리는 ML 파이프라인에 필요한 모든 구성 요소를 제공한다. 파이프라인 업무 간의 취약한 연결(Glue code)은 고장을 일으키고, 자주 업데이트하기 어렵다. 구글이 이를 해결하기 위해서 코드의 양을 최소화하는 플랫폼을 개발하기로 결정했는데, 그게 바로 TFX이다.

TFX는 아래와 같이 다양한 파이프라인 컴포넌트를 제공한다.

TFX Components (출처 : tensorflow.org)

설치는 일반 라이브러리 설치하듯 pip로 설치하면 된다.

$pip install tfx==1.2.0

라이브러리 내의 컴포넌트는 단일 태스크보다 더 복잡한 프로세스를 처리한다.

• 입력 수신 (driver)
• 작업 수행 (executor)
• 최종 결과 저장 (publisher)

컴포넌트의 입력 및 출력을 아티팩트라고하고, 예를 들어 raw data, preprocessed data, model 객체가 있다. 각 아티팩트는 메타테이터스토어에 저장된 메타데이터와 연결된다. TFX 컴포넌트는 메타데이터로 의사소통을 한다. TFX 컴포넌트는 끊임없이 추적하고 관리하기 때문에 MLMD(ML Metadata)는 다양한 기능을 제공할 수 있다.

InteractiveContext 예제

대화형 파이프라인인 컨텍스트 객체를 만들어보자. 먼저 아래의 코드를 통해 InteractiveContext객체를 만든다.

context = InteractiveContext()

run함수와 show함수를 통해 컴포넌트 개체를 실행하고, 메타데이터스토어에 자동으로 기록한다. 주피터 노트북에 출력을 표시하여 즉시 시각화 할 수 있다. 아래의 코드는 StatisticsGen 컴포넌트를 사용하여 데이터셋의 피처를 검사하는 예제이다.

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

주피터 노트북에서의 StatisticsGen 컴포넌트 시각화 예제

Apache Beam

아파치 빔을 사용하면 다양한 TFX컴포넌트와 라이브러리를 처리할 수 있다. 다양한 환경에서 실행할 수 있는 오픈소스이고, 데이터 처리 과정을 추상화하며 여러 분산 처리 런타임환경에서 실행 될 수 있다. 파이프라인을 변경하지 않고 Spark나 Dataflow(GCP)에서 동일한 데이터 파이프라인을 실행할 수 있다. 

 

아래의 코드를 통해 설치한다. (pip로 TFX를 설치했다면 자동으로 설치된다)

$pip install apache-beam
$pip install 'apache-beam[gcp]' #for GCP
$pip install 'apache-beam[boto]' #for AWS

아파치 빔은 컬렉션과 변환이라는 두 개념을 기반으로 한다.

• 컬렉션(PCollections) : 지정된 파일 또는 스트림에서 데이터를 읽거나 쓰는 작업
• 변환(Transform) : 데이터를 조작
• 모든 컬렉션과 변환은 context manager 명령을 통해서 파이썬으로 표시된다.

Pcollections & Transform 예제

shakespeare/kinglear.txt 예제 데이터를 통해서 워드 카운팅하는 작업을 만들어 보았다. 

• 파이프 연산자 | 를 통해 체인으로 연결
• | 와 >> 사이의 문자열 식별자로 표시되는 작업 이름을 제공

pipeline_options = PipelineOptions() #객체 정의

with beam.Pipeline(options=pipeline_options) as p:
    lines = p | ReadFromText(input_file) #파일을 읽어서 데이터 콜렉션을 생성
    
    counts =(
        lines
        | 'Split' >> beam.FlatMap(lambda x: re.findall(r'[A-Za-z\']+', x)) #transform1
        | 'PairWithOne' >> beam.Map(lambda x: (x, 1)) #transform2
        | 'GroupAndSum' >> beam.CombinePerKey(sum)) #transform3
    
    def format_result(word_count):
        (word, count) = word_count
        return "{}:{}".format(word, count) 
    
    output = counts | 'Format' >> beam.Map(format_result) #transform4
    
    output | WriteToText(output_file) #output 저장

결과는 아래와 같다.

output.txt

이상 아파치 빔을 이용해서 데이터 변환을 수행해보았다.