Building ML Pipelines 따라잡기 (5) - 데이터 전처리

머신러닝 모델이 학습을 하려면, 학습할 수 있는 포맷의 데이터를 제공해야 한다(e.g. 카테고리 값들을 수치화). 일관된 전처리는 머신러닝 파이프라인에서 매우 중요한 요소이며, TFX 컴포넌트에서는 어떻게 제공 되는지 알아보자.

TFT(Tensorflow Transform)

TFT를 사용하면 전처리 단계를 텐서플로 그래프로 구성할 수 있다. TFT를 통해서 전처리를 표준화 해야하는 이유는 크게 3가지이다.

1. 전체 데이터셋의 context에서 효율적으로 전처리
   : TFT는 백그라운드에서 데이터를 통과하는 path를 관리하는 피처를 제공한다.
2. 전처리 단계를 효과적으로 확장
   : TFT는 내부적으로 아파치 빔을 사용하여 전처리 알고리즘을 실행하기에, 빔 백엔드에서 전처리를 배포할 수 있다.
3. 잠재적인 학습-서빙 왜곡을 방지
   : 학습 중에 사용한 전처리 단계가 추론 중에 사용한 단계와 일치하지 않으면 학습-서빙 왜곡이 발생한다. 일반적으로 학습 단계에서는 Pandas나 Spark를 통해서 전처리하고, 예측 단계에서는 데이터가 모델에 도달하기 전에 API에서 전처리 단계가 구현되는데, 일치하지 않을 경우가 생긴다. TFT는 이를 단순화 한다.

import pprint
import tempfile

import tensorflow as tf
import tensorflow_transform as tft
import tensorflow_transform.beam as tft_beam
from tensorflow_transform.tf_metadata import dataset_metadata
from tensorflow_transform.tf_metadata import schema_utils


def preprocessing_fn(inputs):
    """Preprocess input columns into transformed columns."""
    print(inputs)
    x = inputs['x']
    x_xf = tft.scale_to_0_1(x)
    return {
        'x_xf': x_xf,
    }

# 데이터 생성
raw_data = [
    {'x': 1.20},
    {'x': 2.99},
    {'x': 100.0}
]

# 데이터 스키마 생성
raw_data_metadata = dataset_metadata.DatasetMetadata(
    schema_utils.schema_from_feature_spec({
        'x': tf.io.FixedLenFeature([], tf.float32),
    }))

# tft_beam.AnalyzeAndTransformDataset(preprocessing_fn) 실행
with tft_beam.Context(temp_dir=tempfile.mkdtemp()):
    transformed_dataset, transform_fn = (  # pylint: disable=unused-variable
            (raw_data, raw_data_metadata) | tft_beam.AnalyzeAndTransformDataset(
        preprocessing_fn))

transformed_data, transformed_metadata = transformed_dataset  # pylint: disable=unused-variable

pprint.pprint(transformed_data)

# >> [{'x_xf': 0.0}, {'x_xf': 0.018117407}, {'x_xf': 1.0}]

AnalyzeAndTransformDataset()는 preprocessing_fn 함수를 적용하고 raw_data, raw_data_metadata 두 인수를 함수에 제공한다. 그런 다음 두 아티팩트를 반환하는데, 전처리된 데이터셋과 함수이다. 반환된 함수는 transform_fn이며 데이터셋에 적용된 변환작업을 나타낸다.

 

예제 실습

먼저 StatisticGen과 SchemaGen을 통해 examples과 schema를 각각 생성한다.

statistics_gen = StatisticsGen(examples=example_gen.outputs["examples"])
    
schema_gen = SchemaGen(
    statistics=statistics_gen.outputs["statistics"],
    infer_feature_shape=False,
)

transform 컴포넌트를 사용하려면 preprocessing_fn()가 모듈 파일에 포함되어야 하고, 그 모듈을 transform 함수 내에서 파라미터로 불러와야 한다.

transform = Transform(
    examples=example_gen.outputs["examples"],
    schema=schema_gen.outputs["schema"],
    module_file=os.path.abspath("transform.py"))
context.run(transform)

"transform.py"에 정의된 변환을 적용하고, 이 변환은 TFRecord 데이터 구조로 변환된다. 이 변환된 데이터와 그래프는 다음 모델 학습단계인 Trainer 컴포넌트에서 사용할 수 있다.