TFX 및 Vertex 파이프라인을 사용하여 BigQuery에서 데이터 읽기

TensorFlow.org에서 보기

Google Colab에서 실행

GitHub에서 소스 보기

노트북 다운로드

Google Cloud Vertex AI Workbench에서 실행

이 노트북 기반 가이드에서는 Google Cloud BigQuery 를 ML 모델을 학습시키기 위한 데이터 소스로 사용합니다. ML 파이프라인은 TFX를 사용하여 구성되고 Google Cloud Vertex Pipelines에서 실행됩니다.

이 노트북은 Vertex Pipelines Tutorial용 Simple TFX Pipeline 에서 구축한 TFX 파이프라인을 기반으로 합니다. 해당 튜토리얼을 아직 읽지 않았다면 이 노트북을 계속 진행하기 전에 읽어야 합니다.

BigQuery 는 비즈니스 민첩성을 위해 설계된 서버리스, 확장성이 뛰어나고 비용 효율적인 멀티 클라우드 데이터 웨어하우스입니다. TFX를 사용하여 BigQuery에서 학습 데이터를 읽고 학습 된 모델을 BigQuery에 게시할 수 있습니다.

이 가이드에서는 BigQuery에서 TFX 파이프라인으로 데이터를 읽는 BigQueryExampleGen 구성요소를 사용합니다.

이 메모장은 Google Colab 또는 AI Platform Notebooks 에서 실행하도록 만들어졌습니다. 이들 중 하나를 사용하지 않는 경우 위의 "Google Colab에서 실행" 버튼을 클릭하기만 하면 됩니다.

설정

버텍스 파이프라인을 위한 단순 TFX 파이프라인 튜토리얼을 완료했다면 작동하는 GCP 프로젝트와 GCS 버킷이 있을 것이며 이것이 이 튜토리얼에 필요한 전부입니다. 놓친 경우 사전 자습서를 먼저 읽으십시오.

파이썬 패키지 설치

TFX 및 KFP를 포함한 필수 Python 패키지를 설치하여 ML 파이프라인을 작성하고 Vertex Pipelines에 작업을 제출합니다.

# Use the latest version of pip.
pip install --upgrade pip
pip install --upgrade "tfx[kfp]<2"

런타임을 다시 시작하셨습니까?

Google Colab을 사용하는 경우 위의 셀을 처음 실행할 때 위의 "RESTART RUNTIME" 버튼을 클릭하거나 "런타임 > 런타임 다시 시작..." 메뉴를 사용하여 런타임을 다시 시작해야 합니다. Colab이 패키지를 로드하는 방식 때문입니다.

Colab에 있지 않다면 다음 셀을 사용하여 런타임을 다시 시작할 수 있습니다.

# docs_infra: no_execute
import sys
if not 'google.colab' in sys.modules:
  # Automatically restart kernel after installs
  import IPython
  app = IPython.Application.instance()
  app.kernel.do_shutdown(True)

이 노트북을 사용하려면 Google에 로그인하세요.

Colab에서 이 노트북을 실행하는 경우 사용자 계정으로 인증하세요.

import sys
if 'google.colab' in sys.modules:
  from google.colab import auth
  auth.authenticate_user()

AI Platform Notebooks 를 사용 중인 경우 다음 섹션을 실행하기 전에 다음을 실행하여 Google Cloud에 인증하세요.

gcloud auth login

터미널 창 에서(메뉴에서 파일 > 새로 만들기를 통해 열 수 있음). 이 작업은 노트북 인스턴스당 한 번만 수행하면 됩니다.

패키지 버전을 확인하십시오.

import tensorflow as tf
print('TensorFlow version: {}'.format(tf.__version__))
from tfx import v1 as tfx
print('TFX version: {}'.format(tfx.__version__))
import kfp
print('KFP version: {}'.format(kfp.__version__))

TensorFlow version: 2.7.1
TFX version: 1.6.0
KFP version: 1.8.11

변수 설정

아래에서 파이프라인을 사용자 지정하는 데 사용되는 몇 가지 변수를 설정합니다. 다음 정보가 필요합니다.

• GCP 프로젝트 ID 및 번호입니다. 프로젝트 ID 및 번호 식별을 참조하십시오.
• 파이프라인을 실행할 GCP 리전 Vertex Pipelines를 사용할 수 있는 지역에 대한 자세한 내용은 Vertex AI 위치 가이드 를 참조하십시오.
• 파이프라인 출력을 저장하기 위한 Google Cloud Storage 버킷.

실행하기 전에 아래 셀에 필요한 값을 입력하십시오 .

GOOGLE_CLOUD_PROJECT = ''         # <--- ENTER THIS
GOOGLE_CLOUD_PROJECT_NUMBER = ''  # <--- ENTER THIS
GOOGLE_CLOUD_REGION = ''          # <--- ENTER THIS
GCS_BUCKET_NAME = ''              # <--- ENTER THIS

if not (GOOGLE_CLOUD_PROJECT and  GOOGLE_CLOUD_PROJECT_NUMBER and GOOGLE_CLOUD_REGION and GCS_BUCKET_NAME):
    from absl import logging
    logging.error('Please set all required parameters.')

ERROR:absl:Please set all required parameters.

프로젝트를 사용하도록 gcloud 를 설정합니다.

gcloud config set project {GOOGLE_CLOUD_PROJECT}

ERROR: (gcloud.config.set) argument VALUE: Must be specified.
Usage: gcloud config set SECTION/PROPERTY VALUE [optional flags]
  optional flags may be  --help | --installation

For detailed information on this command and its flags, run:
  gcloud config set --help

PIPELINE_NAME = 'penguin-bigquery'

# Path to various pipeline artifact.
PIPELINE_ROOT = 'gs://{}/pipeline_root/{}'.format(
    GCS_BUCKET_NAME, PIPELINE_NAME)

# Paths for users' Python module.
MODULE_ROOT = 'gs://{}/pipeline_module/{}'.format(
    GCS_BUCKET_NAME, PIPELINE_NAME)

# Paths for users' data.
DATA_ROOT = 'gs://{}/data/{}'.format(GCS_BUCKET_NAME, PIPELINE_NAME)

# This is the path where your model will be pushed for serving.
SERVING_MODEL_DIR = 'gs://{}/serving_model/{}'.format(
    GCS_BUCKET_NAME, PIPELINE_NAME)

print('PIPELINE_ROOT: {}'.format(PIPELINE_ROOT))

PIPELINE_ROOT: gs:///pipeline_root/penguin-bigquery

기본적으로 Vertex Pipelines는 [project-number]-compute@developer.gserviceaccount.com 형식의 기본 GCE VM 서비스 계정을 사용합니다. 파이프라인에서 BigQuery에 액세스하려면 이 계정에 BigQuery를 사용할 수 있는 권한을 부여해야 합니다. 계정에 'BigQuery 사용자' 역할을 추가합니다.

!gcloud projects add-iam-policy-binding {GOOGLE_CLOUD_PROJECT} \
  --member=serviceAccount:{GOOGLE_CLOUD_PROJECT_NUMBER}-compute@developer.gserviceaccount.com \
  --role=roles/bigquery.user

ERROR: (gcloud.projects.add-iam-policy-binding) argument PROJECT_ID: Must be specified.
Usage: gcloud projects add-iam-policy-binding PROJECT_ID --member=PRINCIPAL --role=ROLE [optional flags]
  optional flags may be  --condition | --condition-from-file | --help

For detailed information on this command and its flags, run:
  gcloud projects add-iam-policy-binding --help

서비스 계정 및 IAM 구성에 대한 자세한 내용은 Vertex 설명서 를 참조하십시오.

파이프라인 생성

TFX 파이프라인 은 정점 파이프라인 자습서용 단순 TFX 파이프라인 에서 했던 것처럼 Python API를 사용하여 정의됩니다. 우리는 이전에 CSV 파일에서 데이터를 읽는 CsvExampleGen 을 사용했습니다. 이 가이드에서는 BigQuery에서 데이터를 읽는 BigQueryExampleGen 구성요소를 사용합니다.

BigQuery 쿼리 준비

우리는 동일한 Palmer Penguins 데이터셋 을 사용할 것입니다. 그러나 동일한 CSV 파일을 사용하여 채워진 BigQuery 테이블 tfx-oss-public.palmer_penguins.palmer_penguins 에서 이를 읽습니다.

Google Colab을 사용하는 경우 BigQuery 테이블의 내용을 직접 확인할 수 있습니다.

# docs_infra: no_execute
%%bigquery --project {GOOGLE_CLOUD_PROJECT}
SELECT *
FROM `tfx-oss-public.palmer_penguins.palmer_penguins`
LIMIT 5

레이블인 species 을 제외한 모든 기능은 이미 0~1로 정규화되었습니다. 우리는 펭귄의 species 을 예측하는 분류 모델을 구축할 것입니다.

BigQueryExampleGen 에는 가져올 데이터를 지정하는 쿼리가 필요합니다. 테이블에 있는 모든 행의 모든 ​​필드를 사용하기 때문에 쿼리는 매우 간단합니다. BigQuery 표준 SQL 구문 에 따라 필요에 따라 필드 이름을 지정하고 WHERE 조건을 추가할 수도 있습니다.

QUERY = "SELECT * FROM `tfx-oss-public.palmer_penguins.palmer_penguins`"

모델 코드를 작성합니다.

Simple TFX Pipeline Tutorial 에서와 동일한 모델 코드를 사용합니다.

_trainer_module_file = 'penguin_trainer.py'

%%writefile {_trainer_module_file}

# Copied from https://www.tensorflow.org/tfx/tutorials/tfx/penguin_simple

from typing import List
from absl import logging
import tensorflow as tf
from tensorflow import keras
from tensorflow_transform.tf_metadata import schema_utils

from tfx import v1 as tfx
from tfx_bsl.public import tfxio

from tensorflow_metadata.proto.v0 import schema_pb2

_FEATURE_KEYS = [
    'culmen_length_mm', 'culmen_depth_mm', 'flipper_length_mm', 'body_mass_g'
]
_LABEL_KEY = 'species'

_TRAIN_BATCH_SIZE = 20
_EVAL_BATCH_SIZE = 10

# Since we're not generating or creating a schema, we will instead create
# a feature spec.  Since there are a fairly small number of features this is
# manageable for this dataset.
_FEATURE_SPEC = {
    **{
        feature: tf.io.FixedLenFeature(shape=[1], dtype=tf.float32)
           for feature in _FEATURE_KEYS
       },
    _LABEL_KEY: tf.io.FixedLenFeature(shape=[1], dtype=tf.int64)
}


def _input_fn(file_pattern: List[str],
              data_accessor: tfx.components.DataAccessor,
              schema: schema_pb2.Schema,
              batch_size: int) -> tf.data.Dataset:
  """Generates features and label for training.

  Args:
    file_pattern: List of paths or patterns of input tfrecord files.
    data_accessor: DataAccessor for converting input to RecordBatch.
    schema: schema of the input data.
    batch_size: representing the number of consecutive elements of returned
      dataset to combine in a single batch

  Returns:
    A dataset that contains (features, indices) tuple where features is a
      dictionary of Tensors, and indices is a single Tensor of label indices.
  """
  return data_accessor.tf_dataset_factory(
      file_pattern,
      tfxio.TensorFlowDatasetOptions(
          batch_size=batch_size, label_key=_LABEL_KEY),
      schema=schema).repeat()


def _make_keras_model() -> tf.keras.Model:
  """Creates a DNN Keras model for classifying penguin data.

  Returns:
    A Keras Model.
  """
  # The model below is built with Functional API, please refer to
  # https://www.tensorflow.org/guide/keras/overview for all API options.
  inputs = [keras.layers.Input(shape=(1,), name=f) for f in _FEATURE_KEYS]
  d = keras.layers.concatenate(inputs)
  for _ in range(2):
    d = keras.layers.Dense(8, activation='relu')(d)
  outputs = keras.layers.Dense(3)(d)

  model = keras.Model(inputs=inputs, outputs=outputs)
  model.compile(
      optimizer=keras.optimizers.Adam(1e-2),
      loss=tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True),
      metrics=[keras.metrics.SparseCategoricalAccuracy()])

  model.summary(print_fn=logging.info)
  return model


# TFX Trainer will call this function.
def run_fn(fn_args: tfx.components.FnArgs):
  """Train the model based on given args.

  Args:
    fn_args: Holds args used to train the model as name/value pairs.
  """

  # This schema is usually either an output of SchemaGen or a manually-curated
  # version provided by pipeline author. A schema can also derived from TFT
  # graph if a Transform component is used. In the case when either is missing,
  # `schema_from_feature_spec` could be used to generate schema from very simple
  # feature_spec, but the schema returned would be very primitive.
  schema = schema_utils.schema_from_feature_spec(_FEATURE_SPEC)

  train_dataset = _input_fn(
      fn_args.train_files,
      fn_args.data_accessor,
      schema,
      batch_size=_TRAIN_BATCH_SIZE)
  eval_dataset = _input_fn(
      fn_args.eval_files,
      fn_args.data_accessor,
      schema,
      batch_size=_EVAL_BATCH_SIZE)

  model = _make_keras_model()
  model.fit(
      train_dataset,
      steps_per_epoch=fn_args.train_steps,
      validation_data=eval_dataset,
      validation_steps=fn_args.eval_steps)

  # The result of the training should be saved in `fn_args.serving_model_dir`
  # directory.
  model.save(fn_args.serving_model_dir, save_format='tf')

Writing penguin_trainer.py

파이프라인 구성 요소에서 액세스할 수 있는 GCS에 모듈 파일을 복사합니다. 모델 학습은 GCP에서 이루어지므로 이 모델 정의를 업로드해야 합니다.

그렇지 않으면 모듈 파일을 포함하는 컨테이너 이미지를 빌드하고 이미지를 사용하여 파이프라인을 실행할 수 있습니다.

gsutil cp {_trainer_module_file} {MODULE_ROOT}/

InvalidUrlError: Cloud URL scheme should be followed by colon and two slashes: "://". Found: "gs:///pipeline_module/penguin-bigquery/".

파이프라인 정의 작성

TFX 파이프라인을 생성하는 함수를 정의합니다. query 를 인수로 사용하는 BigQueryExampleGen 을 사용해야 합니다. 이전 튜토리얼에서 한 가지 더 변경된 점은 컴포넌트가 실행될 때 컴포넌트에 전달되는 beam_pipeline_args 를 전달해야 한다는 것입니다. beam_pipeline_args 를 사용하여 BigQuery에 추가 매개변수를 전달합니다.

from typing import List, Optional

def _create_pipeline(pipeline_name: str, pipeline_root: str, query: str,
                     module_file: str, serving_model_dir: str,
                     beam_pipeline_args: Optional[List[str]],
                     ) -> tfx.dsl.Pipeline:
  """Creates a TFX pipeline using BigQuery."""

  # NEW: Query data in BigQuery as a data source.
  example_gen = tfx.extensions.google_cloud_big_query.BigQueryExampleGen(
      query=query)

  # Uses user-provided Python function that trains a model.
  trainer = tfx.components.Trainer(
      module_file=module_file,
      examples=example_gen.outputs['examples'],
      train_args=tfx.proto.TrainArgs(num_steps=100),
      eval_args=tfx.proto.EvalArgs(num_steps=5))

  # Pushes the model to a file destination.
  pusher = tfx.components.Pusher(
      model=trainer.outputs['model'],
      push_destination=tfx.proto.PushDestination(
          filesystem=tfx.proto.PushDestination.Filesystem(
              base_directory=serving_model_dir)))

  components = [
      example_gen,
      trainer,
      pusher,
  ]

  return tfx.dsl.Pipeline(
      pipeline_name=pipeline_name,
      pipeline_root=pipeline_root,
      components=components,
      # NEW: `beam_pipeline_args` is required to use BigQueryExampleGen.
      beam_pipeline_args=beam_pipeline_args)

정점 파이프라인에서 파이프라인을 실행합니다.

Vertex Pipelines 튜토리얼을 위한 Simple TFX Pipeline 에서 했던 것처럼 Vertex Pipelines를 사용하여 파이프라인을 실행할 것입니다.

또한 BigQueryExampleGen에 대해 beam_pipeline_args 를 전달해야 합니다. 여기에는 GCP 프로젝트 이름 및 BigQuery 실행을 위한 임시 저장소와 같은 구성이 포함됩니다.

import os

# We need to pass some GCP related configs to BigQuery. This is currently done
# using `beam_pipeline_args` parameter.
BIG_QUERY_WITH_DIRECT_RUNNER_BEAM_PIPELINE_ARGS = [
   '--project=' + GOOGLE_CLOUD_PROJECT,
   '--temp_location=' + os.path.join('gs://', GCS_BUCKET_NAME, 'tmp'),
   ]

PIPELINE_DEFINITION_FILE = PIPELINE_NAME + '_pipeline.json'

runner = tfx.orchestration.experimental.KubeflowV2DagRunner(
    config=tfx.orchestration.experimental.KubeflowV2DagRunnerConfig(),
    output_filename=PIPELINE_DEFINITION_FILE)
_ = runner.run(
    _create_pipeline(
        pipeline_name=PIPELINE_NAME,
        pipeline_root=PIPELINE_ROOT,
        query=QUERY,
        module_file=os.path.join(MODULE_ROOT, _trainer_module_file),
        serving_model_dir=SERVING_MODEL_DIR,
        beam_pipeline_args=BIG_QUERY_WITH_DIRECT_RUNNER_BEAM_PIPELINE_ARGS))

생성된 정의 파일은 kfp 클라이언트를 사용하여 제출할 수 있습니다.

# docs_infra: no_execute
from google.cloud import aiplatform
from google.cloud.aiplatform import pipeline_jobs

aiplatform.init(project=GOOGLE_CLOUD_PROJECT, location=GOOGLE_CLOUD_REGION)

job = pipeline_jobs.PipelineJob(template_path=PIPELINE_DEFINITION_FILE,
                                display_name=PIPELINE_NAME)
job.run(sync=False)

이제 Google Cloud Console 에서 'Vertex AI > Pipelines'를 방문하여 진행 상황을 확인할 수 있습니다.