Book a Demo!
CoCalc Logo Icon
StoreFeaturesDocsShareSupportNewsAboutPoliciesSign UpSign In
tensorflow
GitHub Repository: tensorflow/docs-l10n
 Path: blob/master/site/ko/tfx/tutorials/transform/simple.ipynb
25118 views
Kernel: Python 3
#@title Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

TensorFlow Transform으로 데이터 전처리하기

TensorFlow Extended(TFX)의 특성 엔지니어링 구성 요소

참고: 설정이 필요하지 않은 Colab 노트북에서 이 튜토리얼을 실행하는 것이 좋습니다! "Google Colab에서 실행"을 클릭하기만 하면 됩니다.

이 예제 colab 노트북은 TensorFlow Transform(tf.Transform)을 사용하여 프로덕션에서 모델을 훈련하고 추론을 제공하는 데 정확히 동일한 코드를 사용하여 데이터를 사전 처리하는 방법에 대한 매우 간단한 예를 제공합니다.

TensorFlow Transform은 훈련 데이터세트에 대한 전체 전달이 필요한 특성 생성을 포함하여 TensorFlow에 대한 입력 데이터를 전처리하기 위한 라이브러리입니다. 예를 들어 TensorFlow Transform을 사용하여 다음을 수행할 수 있습니다.

  • 평균과 표준 편차를 이용하여 입력값 정규화

  • 모든 입력값에 대해 어휘를 생성하여 문자열을 정수로 변환

  • 관찰된 데이터 분포를 기반으로 부동 소수점을 버킷에 할당하여 정수로 변환

TensorFlow는 단일 예제 또는 예제 배치에 대한 조작을 기본적으로 지원합니다. tf.Transform은 이러한 기능을 확장하여 전체 훈련 데이터세트에 대한 전체 전달을 지원합니다.

tf.Transform의 출력은 훈련과 제공 모두에 사용할 수 있는 TensorFlow 그래프로 내보내집니다. 훈련과 제공 모두에 동일한 그래프를 사용하면 두 단계에 동일한 변환이 적용되므로 왜곡을 방지할 수 있습니다.

Pip 업그레이드

로컬에서 실행할 때 시스템에서 Pip을 업그레이드하지 않으려면 Colab에서 실행 중인지 확인하세요. 물론 로컬 시스템은 별도로 업그레이드할 수 있습니다.

try:
  import colab
  !pip install --upgrade pip
except:
  pass

TensorFlow Transform 설치하기

!pip install -q -U tensorflow_transform

# This cell is only necessary because packages were installed while python was
# running. It avoids the need to restart the runtime when running in Colab.
import pkg_resources
import importlib

importlib.reload(pkg_resources)

가져오기

import pathlib
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

데이터: 일부 더미 데이터 생성

간단한 예제를 위해 몇 가지 간단한 더미 데이터를 만들 것입니다.

  • raw_data는 전처리할 초기 원시 데이터입니다.

  • raw_data_metadata에는 raw_data의 각 컬럼 유형을 알려주는 스키마가 포함됩니다. 이 경우에는 매우 간단합니다.

raw_data = [
      {'x': 1, 'y': 1, 's': 'hello'},
      {'x': 2, 'y': 2, 's': 'world'},
      {'x': 3, 'y': 3, 's': 'hello'}
  ]

raw_data_metadata = dataset_metadata.DatasetMetadata(
    schema_utils.schema_from_feature_spec({
        'y': tf.io.FixedLenFeature([], tf.float32),
        'x': tf.io.FixedLenFeature([], tf.float32),
        's': tf.io.FixedLenFeature([], tf.string),
    }))

변환: 전처리 함수 생성

전처리 함수는 tf.Transform의 가장 중요한 개념입니다. 전처리 함수는 데이터세트의 변환이 실제로 일어나는 곳으로, 텐서 사전을 수락하고 반환합니다. 여기서 텐서는 Tensor 또는 SparseTensor를 의미합니다. 일반적으로 전처리 함수의 핵심을 구성하는 두 가지 주요 API 호출 그룹이 있습니다.

  1. TensorFlow Ops: 일반적으로 TensorFlow 연산을 의미하는 텐서를 수락하고 반환하는 모든 함수로, 원시 데이터를 변환된 데이터로 한 번에 하나의 특성 벡터씩 변환하는 TensorFlow 연산을 그래프에 추가합니다. 이는 훈련 및 제공 기간 동안 모든 예에 대해 실행됩니다.

  2. Tensorflow 변환 분석기/맵퍼: tf.Transform에서 제공하는 모든 분석기/맵퍼입니다. 이들은 또한 텐서를 수락 및 반환하며 일반적으로 Tensorflow 연산과 Beam 계산을 조합적으로 포함하지만 TensorFlow 연산과 달리 전체 훈련 데이터세트에 대한 전체 패스가 필요한 분석 중에 Beam 파이프라인에서만 실행됩니다. Beam 계산은 훈련하기 전과 훈련하는 동안 한 번만 실행되며 일반적으로 전체 훈련 데이터세트에 대해 전체 패스를 만듭니다. 그래프에 추가되는 tf.constant 텐서가 생성됩니다. 예를 들어, tft.min은 훈련 데이터세트에 대해 텐서의 최솟값을 계산합니다.

주의: 전처리 함수를 추론 제공에 적용할 때 훈련 중에 분석기가 생성한 상수는 변경되지 않습니다. 데이터에 추세 또는 계절성 요소가 있는 경우 그에 따라 계획하세요.

참고: preprocessing_fn은 직접 호출할 수 없습니다. 즉, preprocessing_fn(raw_data) 호출이 작동하지 않습니다. 대신 다음 셀과 같이 Transform Beam API에 전달해야 합니다.

def preprocessing_fn(inputs):
    """Preprocess input columns into transformed columns."""
    x = inputs['x']
    y = inputs['y']
    s = inputs['s']
    x_centered = x - tft.mean(x)
    y_normalized = tft.scale_to_0_1(y)
    s_integerized = tft.compute_and_apply_vocabulary(s)
    x_centered_times_y_normalized = (x_centered * y_normalized)
    return {
        'x_centered': x_centered,
        'y_normalized': y_normalized,
        's_integerized': s_integerized,
        'x_centered_times_y_normalized': x_centered_times_y_normalized,
    }

구문

이제 모든 것을 한곳으로 모으고 Apache Beam을 사용하여 실행할 준비가 완료되었습니다.

Apache Beam은 변환을 정의하고 호출하기 위해 특수 구문을 사용합니다. 예를 들어 다음 줄을 보겠습니다.

result = pass_this | 'name this step' >> to_this_call

to_this_call 메서드는 pass_this라는 개체를 호출 및 전달하고 이 연산을 스택 추적에서 name this step이라고 합니다. to_this_call에 대한 호출의 결과는 result에서 반환됩니다. 다음과 같이 함께 연결된 파이프라인의 단계를 종종 볼 수 있습니다.

result = apache_beam.Pipeline() | 'first step' >> do_this_first() | 'second step' >> do_this_last()

그리고 새 파이프라인으로 시작했기 때문에 다음과 같이 계속할 수 있습니다.

next_result = result | 'doing more stuff' >> another_function()

종합적으로 살펴보기

이제 데이터를 변환할 준비가 되었습니다. 직접 러너와 함께 Apache Beam을 사용하고 다음 세 가지 입력을 제공합니다.

  1. raw_data - 위에서 생성한 원시 입력 데이터

  2. raw_data_metadata - 원시 데이터의 스키마

  3. preprocessing_fn - 변환을 수행하기 위해 만든 함수

def main(output_dir):
  # Ignore the warnings
  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

  # Save the transform_fn to the output_dir
  _ = (
      transform_fn
      | 'WriteTransformFn' >> tft_beam.WriteTransformFn(output_dir))

  return transformed_data, transformed_metadata

output_dir = pathlib.Path(tempfile.mkdtemp())

transformed_data, transformed_metadata = main(str(output_dir))

print('\nRaw data:\n{}\n'.format(pprint.pformat(raw_data)))
print('Transformed data:\n{}'.format(pprint.pformat(transformed_data)))

이것이 정답인가요?

이전에는 이를 위해 tf.Transform을 사용했습니다.

x_centered = x - tft.mean(x)
y_normalized = tft.scale_to_0_1(y)
s_integerized = tft.compute_and_apply_vocabulary(s)
x_centered_times_y_normalized = (x_centered * y_normalized)

  • x_centered - [1, 2, 3]을 입력할 경우 x의 평균은 2이며, 이를 x에서 빼서 x값의 중심을 0에 둡니다. 따라서 [-1.0, 0.0, 1.0]의 결과가 정확합니다.

  • y_normalized - 0과 1 사이에서 y 값을 조정하고 싶었습니다. 우리의 입력은 {code 1}[1, 2, 3]{/code 1}이므로 [0.0, 0.5, 1.0]의 결과가 정확합니다.

  • s_integerized - 우리는 문자열을 어휘의 색인에 매핑하고 싶었으나 우리의 어휘에는 2개의 단어("hello"와 "world")만 있었습니다. 따라서 ["hello", "world", "hello"]를 입력하면 [0, 1, 0]의 결과가 정확합니다. "hello"는 이 데이터에서 가장 자주 발생하므로 어휘의 첫 번째 항목이 됩니다.

  • x_centered_times_y_normalized - 우리는 곱셈을 사용하여 x_center와 y_normalized를 교차하는 새로운 기능을 만들고 싶었습니다. 이것은 원래의 값이 아닌 결과를 곱하며 새 결과인 [-0.0, 0.0, 1.0]가 정확합니다.

transform_fn의 결과 사용하기

!ls -l {output_dir}

transform_fn/ 디렉터리에는 그래프에 빌드된 모든 상수 tensorflow-transform 분석 결과로 구현하는 tf.saved_model가 포함되어 있습니다.

tf.saved_model.load로 직접 로드할 수도 있지만 사용이 쉽지 않습니다.

loaded = tf.saved_model.load(str(output_dir/'transform_fn'))
loaded.signatures['serving_default']

더 나은 접근 방식은 tft.TFTransformOutput을 사용하여 로드하는 것입니다. TFTransformOutput.transform_features_layer 메서드는 변환을 적용하는 데 사용할 수 있는 tft.TransformFeaturesLayer 객체를 반환합니다.

tf_transform_output = tft.TFTransformOutput(output_dir)

tft_layer = tf_transform_output.transform_features_layer()
tft_layer

이 tft.TransformFeaturesLayer는 배치 기능의 사전을 예상합니다. 따라서 raw_data의 List[Dict[str, Any]]로부터 Dict[str, tf.Tensor]를 생성합니다.

raw_data_batch = {
    's': tf.constant([ex['s'] for ex in raw_data]),
    'x': tf.constant([ex['x'] for ex in raw_data], dtype=tf.float32),
    'y': tf.constant([ex['y'] for ex in raw_data], dtype=tf.float32),
}

자체적으로 tft.TransformFeaturesLayer를 사용할 수 있습니다.

transformed_batch = tft_layer(raw_data_batch)

{key: value.numpy() for key, value in transformed_batch.items()}

내보내기

더 일반적인 사용 사례는 tf.Transform을 사용하여 훈련 및 평가 데이터세트에 변환을 적용하는 것입니다(예시는 다음 튜토리얼을 참조). 그런 다음 훈련 후 모델을 내보내기 전에 tft.TransformFeaturesLayer를 첫 번째 레이어로 첨부하여 tf.saved_model의 일부로 내보낼 수 있도록 합니다. 구체적인 예제를 확인하려면 계속 읽으세요.

예제 훈련 모델

아래의 모델은,

  1. 변환된 배치를 취합니다.

  2. 변환된 배치를 하나로 모아 간단한 (batch, features) 행렬로 만듭니다.

  3. 이를 몇 개의 밀집 레이어를 통해 실행합니다.

  4. 10개의 리니어 출력을 생성합니다.

실제 사용 사례에서는 원-핫을 s_integerized 기능에 적용합니다.

tf.Transform으로 변환한 데이터세트에서 이 모델을 훈련할 수 있습니다.

class StackDict(tf.keras.layers.Layer):
  def call(self, inputs):
    values = [
        tf.cast(v, tf.float32)
        for k,v in sorted(inputs.items(), key=lambda kv: kv[0])]
    return tf.stack(values, axis=1)

class TrainedModel(tf.keras.Model):
  def __init__(self):
    super().__init__(self)
    self.concat = StackDict()
    self.body = tf.keras.Sequential([
        tf.keras.layers.Dense(64, activation='relu'),
        tf.keras.layers.Dense(64, activation='relu'),
        tf.keras.layers.Dense(10),
    ])

  def call(self, inputs, training=None):
    x = self.concat(inputs)
    return self.body(x, training)

trained_model = TrainedModel()

우리가 모델을 훈련했다고 상상해 봅니다.

trained_model.compile(loss=..., optimizer='adam')
trained_model.fit(...)

이 모델은 변환된 입력해서 실행됩니다.

trained_model_output = trained_model(transformed_batch)
trained_model_output.shape

내보내기 예제 래퍼

위의 모델을 훈련하고 내보내고 싶어하는 경우를 생각해 봅니다.

이 경우 여러분은 내보낸 모델에 변환 함수를 포함하고 싶어할 것입니다.

class ExportModel(tf.Module):
  def __init__(self, trained_model, input_transform):
    self.trained_model = trained_model
    self.input_transform = input_transform

  @tf.function
  def __call__(self, inputs, training=None):
    x = self.input_transform(inputs)
    return self.trained_model(x)

export_model = ExportModel(trained_model=trained_model,
                           input_transform=tft_layer)

이 결합 모델은 원시 데이터에서 작동하며 훈련된 모델을 직접 호출하는 것과 정확히 동일한 결과를 생성합니다.

export_model_output = export_model(raw_data_batch)
export_model_output.shape

tf.reduce_max(abs(export_model_output - trained_model_output)).numpy()

이 export_model은 tft.TransformFeaturesLayer를 포함하며 완전히 독립적입니다. 이를 저장하고 다른 환경에서 복원해도 똑같은 결과를 얻을 수 있습니다.

import tempfile
model_dir = tempfile.mkdtemp(suffix='tft')

tf.saved_model.save(export_model, model_dir)

reloaded = tf.saved_model.load(model_dir)

reloaded_model_output = reloaded(raw_data_batch)
reloaded_model_output.shape

tf.reduce_max(abs(export_model_output - reloaded_model_output)).numpy()