Yerel Olarak TFX Boru Hattı Oluşturmak

TFX, aşağıdakileri gerçekleştirmek için makine öğrenimi (ML) iş akışınızı bir ardışık düzen olarak düzenlemenizi kolaylaştırır:

  • Modelinizi düzenli olarak yeniden eğitmenize, değerlendirmenize ve dağıtmanıza olanak tanıyan makine öğrenimi sürecinizi otomatikleştirin.
  • Performans ve güvenilirliği sağlamak için model performansının derinlemesine analizini ve yeni eğitilen modellerin doğrulanmasını içeren ML ardışık düzenleri oluşturun.
  • Anormallikler için eğitim verilerini izleyin ve eğitim sunumunda çarpıklığı ortadan kaldırın
  • Farklı hiperparametre kümeleriyle bir işlem hattı çalıştırarak deneme hızını artırın.

Tipik bir üretim hattı geliştirme süreci, üretime geçmeden önce yerel bir makinede veri analizi ve bileşen kurulumuyla başlar. Bu kılavuzda yerel olarak bir boru hattı oluşturmanın iki yolu açıklanmaktadır.

  • TFX işlem hattı şablonunu makine öğrenimi iş akışınızın ihtiyaçlarına uyacak şekilde özelleştirin. TFX işlem hattı şablonları, TFX standart bileşenlerini kullanan en iyi uygulamaları gösteren önceden oluşturulmuş iş akışlarıdır.
  • TFX'i kullanarak bir işlem hattı oluşturun. Bu kullanım durumunda, bir şablondan başlamadan bir işlem hattını tanımlarsınız.

İşlem hattınızı geliştirirken onu LocalDagRunner ile çalıştırabilirsiniz. Daha sonra boru hattı bileşenleri iyi tanımlanıp test edildikten sonra Kubeflow veya Airflow gibi üretim düzeyinde bir orkestratör kullanırsınız.

Başlamadan önce

TFX bir Python paketi olduğundan sanal ortam veya Docker konteyneri gibi bir Python geliştirme ortamı kurmanız gerekecektir. Daha sonra:

pip install tfx

TFX işlem hatları konusunda yeniyseniz devam etmeden önce TFX işlem hatlarına ilişkin temel kavramlar hakkında daha fazla bilgi edinin .

Şablon kullanarak işlem hattı oluşturma

TFX İşlem Hattı Şablonları, kullanım durumunuza göre özelleştirebileceğiniz önceden oluşturulmuş bir işlem hattı tanımları seti sağlayarak işlem hattı geliştirmeye başlamayı kolaylaştırır.

Aşağıdaki bölümlerde bir şablonun kopyasının nasıl oluşturulacağı ve ihtiyaçlarınıza göre nasıl özelleştirileceği açıklanmaktadır.

İşlem hattı şablonunun bir kopyasını oluşturun

  1. Mevcut TFX işlem hattı şablonlarının listesine bakın:

    tfx template list

  2. Listeden bir şablon seçin

    tfx template copy --model=template --pipeline_name=pipeline-name \
--destination_path=destination-path

    Aşağıdakileri değiştirin:

    • template : Kopyalamak istediğiniz şablonun adı.
    • pipeline-name : Oluşturulacak ardışık düzenin adı.
    • destination-path : Şablonun kopyalanacağı yol.

    tfx template copy komutu hakkında daha fazla bilgi edinin.

  3. Belirttiğiniz yolda ardışık düzen şablonunun bir kopyası oluşturuldu.

İşlem hattı şablonunu keşfedin

Bu bölüm, bir şablon tarafından oluşturulan iskeleye genel bir bakış sağlar.

  1. İşlem hattınızın kök dizinine kopyalanan dizinleri ve dosyaları keşfedin

    • Bir boru hattı dizini
      • pipeline.py - boru hattını tanımlar ve hangi bileşenlerin kullanıldığını listeler
      • configs.py - verilerin nereden geldiği veya hangi orkestratörün kullanıldığı gibi yapılandırma ayrıntılarını tutar
    • Bir veri dizini
      • Bu genellikle ExampleGen için varsayılan kaynak olan bir data.csv dosyasını içerir. Veri kaynağını configs.py dosyasında değiştirebilirsiniz.

    • Ön işleme kodu ve model uygulamalarına sahip bir modeller dizini

    • Şablon, yerel ortam ve Kubeflow için DAG çalıştırıcılarını kopyalar.

    • Bazı şablonlar, verilerinizi ve yapıtlarınızı Machine Learning MetaData ile keşfedebilmeniz için Python Not Defterlerini de içerir.

  2. İşlem hattı dizininizde aşağıdaki komutları çalıştırın: 

    tfx pipeline create --pipeline_path local_runner.py

    tfx run create --pipeline_name pipeline_name

    Komut, LocalDagRunner kullanarak aşağıdaki dizinleri ardışık düzeninize ekleyen bir ardışık düzen çalıştırması oluşturur:

    • Yerel olarak kullanılan ML Meta Veri deposunu içeren bir tfx_metadata dizini.
    • İşlem hattının dosya çıktılarını içeren bir tfx_pipeline_output dizini.

  3. İşlem hattınızın pipeline/configs.py dosyasını açın ve içeriğini inceleyin. Bu komut dosyası, işlem hattı ve bileşen işlevleri tarafından kullanılan yapılandırma seçeneklerini tanımlar. Burası, veri kaynağının konumu veya bir çalıştırmadaki eğitim adımlarının sayısı gibi şeyleri belirteceğiniz yerdir.

  4. İşlem hattınızın pipeline/pipeline.py dosyasını açın ve içeriğini inceleyin. Bu komut dosyası TFX işlem hattını oluşturur. Başlangıçta işlem hattı yalnızca bir ExampleGen bileşeni içerir.

    • İşlem hattına daha fazla adım eklemek için pipeline.py dosyasındaki TODO yorumlarındaki talimatları izleyin.

  5. local_runner.py dosyasını açın ve içeriğini inceleyin. Bu betik bir işlem hattı çalıştırması oluşturur ve çalıştırmanın data_path ve preprocessing_fn gibi parametrelerini belirtir.

  6. Şablon tarafından oluşturulan yapı iskelesini incelediniz ve LocalDagRunner kullanarak bir işlem hattı çalıştırması oluşturdunuz. Daha sonra şablonu gereksinimlerinize uyacak şekilde özelleştirin.

İşlem hattınızı özelleştirin

Bu bölümde şablonunuzu özelleştirmeye nasıl başlayacağınıza ilişkin bir genel bakış sunulmaktadır.

  1. Boru hattınızı tasarlayın. Bir şablonun sağladığı yapı iskelesi, TFX standart bileşenlerini kullanarak tablolu veriler için bir işlem hattı uygulamanıza yardımcı olur. Mevcut bir makine öğrenimi iş akışını bir ardışık düzene taşıyorsanız, TFX standart bileşenlerinden tam olarak yararlanmak için kodunuzu gözden geçirmeniz gerekebilir. Ayrıca, iş akışınıza özgü veya henüz TFX standart bileşenleri tarafından desteklenmeyen özellikleri uygulayan özel bileşenler de oluşturmanız gerekebilir.

  2. İşlem hattınızı tasarladıktan sonra, aşağıdaki işlemi kullanarak işlem hattını yinelemeli olarak özelleştirin. Verileri ardışık düzeninize alan bileşenden başlayın; bu genellikle ExampleGen bileşenidir.

    1. İşlem hattını veya bir bileşeni kullanım durumunuza uyacak şekilde özelleştirin. Bu özelleştirmeler aşağıdaki gibi değişiklikleri içerebilir:

      • Boru hattı parametrelerini değiştirme.
      • Boru hattına bileşenler ekleme veya bunları kaldırma.
      • Veri giriş kaynağının değiştirilmesi. Bu veri kaynağı bir dosya veya BigQuery gibi hizmetlere yapılan sorgular olabilir.
      • İşlem hattındaki bir bileşenin yapılandırmasını değiştirme.
      • Bir bileşenin kişiselleştirme fonksiyonunun değiştirilmesi.

    2. local_runner.py betiğini veya farklı bir orkestratör kullanıyorsanız başka bir uygun DAG çalıştırıcısını kullanarak bileşeni yerel olarak çalıştırın. Betik başarısız olursa, hatanın hatalarını ayıklayın ve betiği çalıştırmayı yeniden deneyin.

    3. Bu özelleştirme işe yaradıktan sonra bir sonraki özelleştirmeye geçin.

  3. Tekrarlanarak çalışarak şablon iş akışındaki her adımı ihtiyaçlarınızı karşılayacak şekilde özelleştirebilirsiniz.

Özel bir işlem hattı oluşturun

Şablon kullanmadan özel bir işlem hattı oluşturma hakkında daha fazla bilgi edinmek için aşağıdaki talimatları kullanın.

  1. Boru hattınızı tasarlayın. TFX standart bileşenleri, eksiksiz bir makine öğrenimi iş akışı uygulamanıza yardımcı olacak kanıtlanmış işlevsellik sağlar. Mevcut bir makine öğrenimi iş akışını bir ardışık düzene taşıyorsanız, TFX standart bileşenlerinden tam olarak yararlanmak için kodunuzu gözden geçirmeniz gerekebilir. Veri artırma gibi özellikleri uygulayan özel bileşenler de oluşturmanız gerekebilir.

  2. Aşağıdaki örneği kullanarak işlem hattınızı tanımlamak için bir komut dosyası oluşturun. Bu kılavuzda bu dosyaya my_pipeline.py adı verilmektedir.

    import os
from typing import Optional, Text, List
from absl import logging
from ml_metadata.proto import metadata_store_pb2
import tfx.v1 as tfx

PIPELINE_NAME = 'my_pipeline'
PIPELINE_ROOT = os.path.join('.', 'my_pipeline_output')
METADATA_PATH = os.path.join('.', 'tfx_metadata', PIPELINE_NAME, 'metadata.db')
ENABLE_CACHE = True

def create_pipeline(
  pipeline_name: Text,
  pipeline_root:Text,
  enable_cache: bool,
  metadata_connection_config: Optional[
    metadata_store_pb2.ConnectionConfig] = None,
  beam_pipeline_args: Optional[List[Text]] = None
):
  components = []

  return tfx.dsl.Pipeline(
        pipeline_name=pipeline_name,
        pipeline_root=pipeline_root,
        components=components,
        enable_cache=enable_cache,
        metadata_connection_config=metadata_connection_config,
        beam_pipeline_args=beam_pipeline_args, 
    )

def run_pipeline():
  my_pipeline = create_pipeline(
      pipeline_name=PIPELINE_NAME,
      pipeline_root=PIPELINE_ROOT,
      enable_cache=ENABLE_CACHE,
      metadata_connection_config=tfx.orchestration.metadata.sqlite_metadata_connection_config(METADATA_PATH)
      )

  tfx.orchestration.LocalDagRunner().run(my_pipeline)

if __name__ == '__main__':
  logging.set_verbosity(logging.INFO)
  run_pipeline()

    Sonraki adımlarda, işlem hattınızı create_pipeline tanımlayacak ve işlem hattınızı yerel çalıştırıcıyı kullanarak yerel olarak çalıştıracaksınız.

    Aşağıdaki işlemi kullanarak işlem hattınızı yinelemeli olarak oluşturun.

    1. İşlem hattını veya bir bileşeni kullanım durumunuza uyacak şekilde özelleştirin. Bu özelleştirmeler aşağıdaki gibi değişiklikleri içerebilir:

      • Boru hattı parametrelerini değiştirme.
      • Boru hattına bileşen ekleme veya bunları kaldırma.
      • Bir veri giriş dosyasını değiştirme.
      • İşlem hattındaki bir bileşenin yapılandırmasını değiştirme.
      • Bir bileşenin kişiselleştirme fonksiyonunun değiştirilmesi.

    2. Bileşeni yerel çalıştırıcıyı kullanarak veya betiği doğrudan çalıştırarak yerel olarak çalıştırın. Betik başarısız olursa, hatanın hatalarını ayıklayın ve betiği çalıştırmayı yeniden deneyin.

    3. Bu özelleştirme işe yaradıktan sonra bir sonraki özelleştirmeye geçin.

    İşlem hattınızın iş akışındaki ilk düğümden başlayın; genellikle ilk düğüm, verileri işlem hattınıza alır.

  3. İş akışınızdaki ilk düğümü ardışık düzeninize ekleyin. Bu örnekte işlem hattı, ./data konumundaki bir dizinden bir CSV yüklemek için ExampleGen standart bileşenini kullanır.

    from tfx.components import CsvExampleGen

DATA_PATH = os.path.join('.', 'data')

def create_pipeline(
  pipeline_name: Text,
  pipeline_root:Text,
  data_path: Text,
  enable_cache: bool,
  metadata_connection_config: Optional[
    metadata_store_pb2.ConnectionConfig] = None,
  beam_pipeline_args: Optional[List[Text]] = None
):
  components = []

  example_gen = tfx.components.CsvExampleGen(input_base=data_path)
  components.append(example_gen)

  return tfx.dsl.Pipeline(
        pipeline_name=pipeline_name,
        pipeline_root=pipeline_root,
        components=components,
        enable_cache=enable_cache,
        metadata_connection_config=metadata_connection_config,
        beam_pipeline_args=beam_pipeline_args, 
    )

def run_pipeline():
  my_pipeline = create_pipeline(
    pipeline_name=PIPELINE_NAME,
    pipeline_root=PIPELINE_ROOT,
    data_path=DATA_PATH,
    enable_cache=ENABLE_CACHE,
    metadata_connection_config=tfx.orchestration.metadata.sqlite_metadata_connection_config(METADATA_PATH)
    )

  tfx.orchestration.LocalDagRunner().run(my_pipeline)

    CsvExampleGen belirtilen veri yolundaki CSV'deki verileri kullanarak serileştirilmiş örnek kayıtlar oluşturur. CsvExampleGen bileşeninin input_base parametresini veri köküyle ayarlayarak.

  4. my_pipeline.py ile aynı dizinde bir data dizini oluşturun. data dizinine küçük bir CSV dosyası ekleyin.

  5. my_pipeline.py betiğinizi çalıştırmak için aşağıdaki komutu kullanın.

    python my_pipeline.py

    Sonuç aşağıdakine benzer bir şey olmalıdır:

    INFO:absl:Component CsvExampleGen depends on [].
INFO:absl:Component CsvExampleGen is scheduled.
INFO:absl:Component CsvExampleGen is running.
INFO:absl:Running driver for CsvExampleGen
INFO:absl:MetadataStore with DB connection initialized
INFO:absl:Running executor for CsvExampleGen
INFO:absl:Generating examples.
INFO:absl:Using 1 process(es) for Local pipeline execution.
INFO:absl:Processing input csv data ./data/* to TFExample.
WARNING:root:Couldn't find python-snappy so the implementation of _TFRecordUtil._masked_crc32c is not as fast as it could be.
INFO:absl:Examples generated.
INFO:absl:Running publisher for CsvExampleGen
INFO:absl:MetadataStore with DB connection initialized
INFO:absl:Component CsvExampleGen is finished.

  6. İşlem hattınıza bileşenleri yinelemeli olarak eklemeye devam edin.