एमएल समुदाय दिवस 9 नवंबर है! TensorFlow, JAX से नई जानकारी के लिए हमसे जुड़ें, और अधिक जानें

टीएफएक्स और वर्टेक्स पाइपलाइनों के साथ वर्टेक्स एआई प्रशिक्षण

यह नोटबुक-आधारित ट्यूटोरियल एक टीएफएक्स पाइपलाइन बनाएगा और चलाएगा जो वर्टेक्स एआई प्रशिक्षण सेवा का उपयोग करके एमएल मॉडल को प्रशिक्षित करता है।

इस नोटबुक TFX पाइपलाइन हम में बनाया पर आधारित है वर्टेक्स पाइपलाइन ट्यूटोरियल के लिए सरल TFX पाइपलाइन । यदि आपने अभी तक उस ट्यूटोरियल को नहीं पढ़ा है, तो इस नोटबुक के साथ आगे बढ़ने से पहले आपको इसे पढ़ना चाहिए।

आप ऑटोएमएल का उपयोग करके वर्टेक्स एआई पर मॉडल को प्रशिक्षित कर सकते हैं, या कस्टम प्रशिक्षण का उपयोग कर सकते हैं। कस्टम प्रशिक्षण में, आप अपने प्रशिक्षण कार्यों को सशक्त बनाने, वितरित प्रशिक्षण को सक्षम करने, हाइपरपैरामीटर ट्यूनिंग का उपयोग करने और GPU के साथ त्वरित करने के लिए कई अलग-अलग मशीन प्रकारों का चयन कर सकते हैं।

इस ट्यूटोरियल में, हम TFX पाइपलाइन में एक मॉडल को प्रशिक्षित करने के लिए कस्टम जॉब के साथ वर्टेक्स एआई ट्रेनिंग का उपयोग करेंगे।

इस नोटबुक पर चलाया जा करने का इरादा है गूगल Colab या पर ऐ मंच नोटबुक । यदि आप इनमें से किसी एक का उपयोग नहीं कर रहे हैं, तो आप बस ऊपर "Google Colab में चलाएँ" बटन पर क्लिक कर सकते हैं।

सेट अप

आप पूरी कर ली है वर्टेक्स पाइपलाइन ट्यूटोरियल के लिए सरल TFX पाइपलाइन , आप एक काम कर जीसीपी परियोजना और एक GCS बाल्टी होगा और कहा कि सभी हम इस ट्यूटोरियल के लिए की जरूरत है। यदि आप चूक गए हैं तो कृपया पहले प्रारंभिक ट्यूटोरियल पढ़ें।

पायथन पैकेज स्थापित करें

हम लेखक एमएल पाइपलाइनों के लिए टीएफएक्स और केएफपी सहित आवश्यक पायथन पैकेज स्थापित करेंगे और वर्टेक्स पाइपलाइनों को नौकरियां जमा करेंगे।

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

क्या आपने रनटाइम को पुनरारंभ किया?

यदि आप Google Colab का उपयोग कर रहे हैं, जब आप पहली बार ऊपर सेल चलाते हैं, तो आपको ऊपर "Runtime RUNTIME" बटन पर क्लिक करके या "रनटाइम> रनटाइम को पुनरारंभ करें ..." मेनू का उपयोग करके रनटाइम को पुनरारंभ करना होगा। ऐसा इसलिए है क्योंकि Colab संकुल को लोड करता है।

यदि आप Colab पर नहीं हैं, तो आप निम्न सेल के साथ रनटाइम को पुनरारंभ कर सकते हैं।

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()

आप अगले अनुभाग चल रहा है, चलाकर से पहले पर ऐ मंच नोटबुक, Google मेघ के साथ प्रमाणित कर रहे हैं

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.5.1
TFX version: 1.2.0
KFP version: 1.8.1

चर सेट करें

हम नीचे पाइपलाइनों को अनुकूलित करने के लिए उपयोग किए जाने वाले कुछ चर स्थापित करेंगे। निम्नलिखित जानकारी आवश्यक है:

चलाने से पहले नीचे सेल में आवश्यक मान दर्ज करें।

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

if not (GOOGLE_CLOUD_PROJECT 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-vertex-training'

# 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-vertex-training

उदाहरण डेटा तैयार करें

हम एक ही उपयोग करेगा पामर पेंगुइन डाटासेट के रूप में सरल TFX पाइपलाइन ट्यूटोरियल

इस डेटासेट में चार सांख्यिक विशेषताएं हैं जिन्हें पहले से ही श्रेणी [0,1] के लिए सामान्यीकृत किया गया था। हम एक वर्गीकरण मॉडल जो भविष्यवाणी का निर्माण करेगा species पेंगुइन की।

हमें डेटासेट की अपनी प्रतिलिपि बनाने की आवश्यकता है। क्योंकि TFX ExampleGen एक निर्देशिका से इनपुट पढ़ता है, हमें GCS पर एक निर्देशिका बनाने और डेटासेट की प्रतिलिपि बनाने की आवश्यकता है।

gsutil cp gs://download.tensorflow.org/data/palmer_penguins/penguins_processed.csv {DATA_ROOT}/
InvalidUrlError: Cloud URL scheme should be followed by colon and two slashes: "://". Found: "gs:///data/penguin-vertex-training/".

सीएसवी फ़ाइल पर एक त्वरित नज़र डालें।

gsutil cat {DATA_ROOT}/penguins_processed.csv | head
InvalidUrlError: Cloud URL scheme should be followed by colon and two slashes: "://". Found: "gs:///data/penguin-vertex-training/penguins_processed.csv".

एक पाइपलाइन बनाएं

हमारे पाइप लाइन बहुत पाइपलाइन हम में बनाया के समान होगा वर्टेक्स पाइपलाइन ट्यूटोरियल के लिए सरल TFX पाइपलाइन । पाइपलाइन में तीन घटक होंगे, CsvExampleGen, ट्रेनर और पुशर। लेकिन हम एक विशेष ट्रेनर घटक का उपयोग करेंगे जिसका उपयोग प्रशिक्षण कार्यभार को वर्टेक्स एआई में स्थानांतरित करने के लिए किया जाता है।

TFX एक विशेष प्रदान करता है Trainer वर्टेक्स ऐ प्रशिक्षण सेवा करने के लिए प्रशिक्षण कार्य सबमिट करने, और सभी हम क्या करना है बस उपयोग कर रहा है Trainer विस्तार मॉड्यूल के बजाय मानक में Trainer कुछ आवश्यक जीसीपी पैरामीटर के साथ साथ घटक।

इस ट्यूटोरियल में, हम केवल पहले CPU का उपयोग करके और फिर GPU के साथ Vertex AI ट्रेनिंग जॉब चलाएंगे।

मॉडल कोड लिखें।

मॉडल ही लगभग में मॉडल के समान है सरल TFX पाइपलाइन ट्यूटोरियल

हम जोड़ देगा _get_distribution_strategy() समारोह जो एक बनाता है TensorFlow वितरण रणनीति और उस में प्रयोग किया जाता है run_fn MirroredStrategy उपयोग करने के लिए करता है, तो GPU उपलब्ध है।

_trainer_module_file = 'penguin_trainer.py'
%%writefile {_trainer_module_file}

# Copied from https://www.tensorflow.org/tfx/tutorials/tfx/penguin_simple and
# slightly modified run_fn() to add distribution_strategy.

from typing import List
from absl import logging
import tensorflow as tf
from tensorflow import keras
from tensorflow_metadata.proto.v0 import schema_pb2
from tensorflow_transform.tf_metadata import schema_utils

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

_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


# NEW: Read `use_gpu` from the custom_config of the Trainer.
#      if it uses GPU, enable MirroredStrategy.
def _get_distribution_strategy(fn_args: tfx.components.FnArgs):
  if fn_args.custom_config.get('use_gpu', False):
    logging.info('Using MirroredStrategy with one GPU.')
    return tf.distribute.MirroredStrategy(devices=['device:GPU:0'])
  return None


# 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)

  # NEW: If we have a distribution strategy, build a model in a strategy scope.
  strategy = _get_distribution_strategy(fn_args)
  if strategy is None:
    model = _make_keras_model()
  else:
    with strategy.scope():
      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 में कॉपी करें जिसे पाइपलाइन घटकों से एक्सेस किया जा सकता है।

अन्यथा, आप मॉड्यूल फ़ाइल सहित एक कंटेनर छवि बनाना चाहते हैं और पाइपलाइन और एआई प्लेटफ़ॉर्म प्रशिक्षण कार्य चलाने के लिए छवि का उपयोग कर सकते हैं।

gsutil cp {_trainer_module_file} {MODULE_ROOT}/
InvalidUrlError: Cloud URL scheme should be followed by colon and two slashes: "://". Found: "gs:///pipeline_module/penguin-vertex-training/".

एक पाइपलाइन परिभाषा लिखें

हम TFX पाइपलाइन बनाने के लिए एक फ़ंक्शन को परिभाषित करेंगे। यह में के रूप में ही तीन घटक होते हैं सरल TFX पाइपलाइन ट्यूटोरियल है, लेकिन हम एक का उपयोग Trainer जीसीपी विस्तार मॉड्यूल में घटक।

tfx.extensions.google_cloud_ai_platform.Trainer एक नियमित रूप से की तरह बर्ताव करती है Trainer , लेकिन यह सिर्फ बादल करने के लिए मॉडल के प्रशिक्षण के लिए गणना ले जाता है। यह वर्टेक्स एआई ट्रेनिंग सर्विस में एक कस्टम जॉब लॉन्च करता है और ऑर्केस्ट्रेशन सिस्टम में ट्रेनर कंपोनेंट वर्टेक्स एआई ट्रेनिंग जॉब पूरा होने तक इंतजार करेगा।

def _create_pipeline(pipeline_name: str, pipeline_root: str, data_root: str,
                     module_file: str, serving_model_dir: str, project_id: str,
                     region: str, use_gpu: bool) -> tfx.dsl.Pipeline:
  """Implements the penguin pipeline with TFX."""
  # Brings data into the pipeline or otherwise joins/converts training data.
  example_gen = tfx.components.CsvExampleGen(input_base=data_root)

  # NEW: Configuration for Vertex AI Training.
  # This dictionary will be passed as `CustomJobSpec`.
  vertex_job_spec = {
      'project': project_id,
      'worker_pool_specs': [{
          'machine_spec': {
              'machine_type': 'n1-standard-4',
          },
          'replica_count': 1,
          'container_spec': {
              'image_uri': 'gcr.io/tfx-oss-public/tfx:{}'.format(tfx.__version__),
          },
      }],
  }
  if use_gpu:
    # See https://cloud.google.com/vertex-ai/docs/reference/rest/v1/MachineSpec#acceleratortype
    # for available machine types.
    vertex_job_spec['worker_pool_specs'][0]['machine_spec'].update({
        'accelerator_type': 'NVIDIA_TESLA_K80',
        'accelerator_count': 1
    })

  # Trains a model using Vertex AI Training.
  # NEW: We need to specify a Trainer for GCP with related configs.
  trainer = tfx.extensions.google_cloud_ai_platform.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),
      custom_config={
          tfx.extensions.google_cloud_ai_platform.ENABLE_UCAIP_KEY:
              True,
          tfx.extensions.google_cloud_ai_platform.UCAIP_REGION_KEY:
              region,
          tfx.extensions.google_cloud_ai_platform.TRAINING_ARGS_KEY:
              vertex_job_spec,
          'use_gpu':
              use_gpu,
      })

  # Pushes the model to a filesystem 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)

पाइपलाइन को वर्टेक्स पाइपलाइनों पर चलाएँ।

हम के रूप में हम में किया था पाइप लाइन को चलाने के लिए वर्टेक्स पाइपलाइन का उपयोग करेगा वर्टेक्स पाइपलाइन ट्यूटोरियल के लिए सरल TFX पाइपलाइन

import os

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,
        data_root=DATA_ROOT,
        module_file=os.path.join(MODULE_ROOT, _trainer_module_file),
        serving_model_dir=SERVING_MODEL_DIR,
        project_id=GOOGLE_CLOUD_PROJECT,
        region=GOOGLE_CLOUD_REGION,
        # We will use CPUs only for now.
        use_gpu=False))

उत्पन्न परिभाषा फ़ाइल kfp क्लाइंट का उपयोग करके प्रस्तुत की जा सकती है।

from kfp.v2.google import client

pipelines_client = client.AIPlatformClient(
    project_id=GOOGLE_CLOUD_PROJECT,
    region=GOOGLE_CLOUD_REGION,
)

_ = pipelines_client.create_run_from_job_spec(PIPELINE_DEFINITION_FILE)

अब आप ऊपर या यात्रा 'वर्टेक्स ऐ> पाइपलाइन' में उत्पादन में लिंक पर जा सकते हैं Google क्लाउड कंसोल प्रगति देखने के लिए।

GPU का उपयोग करके पाइपलाइन चलाएं

वर्टेक्स एआई जीपीयू के लिए समर्थन सहित विभिन्न मशीन प्रकारों का उपयोग करके प्रशिक्षण का समर्थन करता है। देखें मशीन कल्पना संदर्भ उपलब्ध विकल्पों के लिए।

हमने GPU प्रशिक्षण का समर्थन करने के लिए अपनी पाइपलाइन को पहले ही परिभाषित कर दिया है। हमें बस की जरूरत है स्थापित करने है use_gpu सही पर झंडा। फिर एक पाइप लाइन के लिए एक मशीन एक NVIDIA_TESLA_K80 और हमारे मॉडल प्रशिक्षण कोड सहित का उपयोग करेगा कल्पना के साथ बनाया जाएगा tf.distribute.MirroredStrategy

ध्यान दें कि use_gpu झंडा वर्टेक्स या TFX एपीआई का एक हिस्सा नहीं है। इसका उपयोग इस ट्यूटोरियल में प्रशिक्षण कोड को नियंत्रित करने के लिए किया जाता है।

runner.run(
    _create_pipeline(
        pipeline_name=PIPELINE_NAME,
        pipeline_root=PIPELINE_ROOT,
        data_root=DATA_ROOT,
        module_file=os.path.join(MODULE_ROOT, _trainer_module_file),
        serving_model_dir=SERVING_MODEL_DIR,
        project_id=GOOGLE_CLOUD_PROJECT,
        region=GOOGLE_CLOUD_REGION,
        # Updated: Use GPUs. We will use a NVIDIA_TESLA_K80 and 
        # the model code will use tf.distribute.MirroredStrategy.
        use_gpu=True))

_ = pipelines_client.create_run_from_job_spec(PIPELINE_DEFINITION_FILE)

अब आप ऊपर या यात्रा 'वर्टेक्स ऐ> पाइपलाइन' में उत्पादन में लिंक पर जा सकते हैं Google क्लाउड कंसोल प्रगति देखने के लिए।