로컬에서 TFX 파이프라인 구축

컬렉션을 사용해 정리하기 내 환경설정을 기준으로 콘텐츠를 저장하고 분류하세요.

TFX를 사용하면 머신 러닝(ML) 워크플로를 파이프라인으로 더 쉽게 오케스트레이션하여 다음을 수행할 수 있습니다.

  • ML 프로세스를 자동화하여 모델을 정기적으로 재교육, 평가 및 배포할 수 있습니다.
  • 성능과 안정성을 보장하기 위해 모델 성능에 대한 심층 분석 및 새로 훈련된 모델의 검증을 포함하는 ML 파이프라인을 생성합니다.
  • 훈련 데이터에서 이상 징후를 모니터링하고 훈련 제공 편향 제거
  • 다양한 하이퍼파라미터 세트로 파이프라인을 실행하여 실험 속도를 높입니다.

일반적인 파이프라인 개발 프로세스는 프로덕션에 배포되기 전에 데이터 분석 및 구성 요소 설정과 함께 로컬 시스템에서 시작됩니다. 이 가이드에서는 로컬에서 파이프라인을 구축하는 두 가지 방법을 설명합니다.

  • ML 워크플로의 요구 사항에 맞게 TFX 파이프라인 템플릿을 사용자 지정합니다. TFX 파이프라인 템플릿은 TFX 표준 구성 요소를 사용하는 모범 사례를 보여 주는 사전 구축된 워크플로입니다.
  • TFX를 사용하여 파이프라인을 빌드합니다. 이 사용 사례에서는 템플릿에서 시작하지 않고 파이프라인을 정의합니다.

당신이 당신의 파이프 라인을 개발하고, 당신은 그것을 실행할 수 있습니다 LocalDagRunner . 그런 다음 파이프라인 구성 요소가 잘 정의되고 테스트되면 Kubeflow 또는 Airflow와 같은 프로덕션 등급 오케스트레이터를 사용합니다.

시작하기 전에

TFX는 Python 패키지이므로 가상 환경이나 Docker 컨테이너와 같은 Python 개발 환경을 설정해야 합니다. 그 다음에:

pip install tfx

당신이 TFX 파이프 라인을 처음 사용하는 경우 더 TFX 파이프 라인에 대한 핵심 개념에 대해 배울 계속하기 전에.

템플릿을 사용하여 파이프라인 구축

TFX 파이프라인 템플릿을 사용하면 사용 사례에 맞게 사용자 지정할 수 있는 미리 빌드된 파이프라인 정의 세트를 제공하여 파이프라인 개발을 더 쉽게 시작할 수 있습니다.

다음 섹션에서는 템플릿 복사본을 만들고 필요에 맞게 사용자 지정하는 방법을 설명합니다.

파이프라인 템플릿 복사본 만들기

  1. 사용 가능한 TFX 파이프라인 템플릿 목록을 참조하십시오.

    tfx template list

  2. 목록에서 템플릿 선택

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

    다음을 교체합니다.

    • template : 복사 할 템플릿의 이름입니다.
    • pipeline-name : 파이프 라인의 이름을 만듭니다.
    • destination-path :로 템플릿을 복사 할 경로입니다.

    [정보 자세히 알아보기 tfx template copy 명령을 .

  3. 지정한 경로에 파이프라인 템플릿의 복사본이 생성되었습니다.

파이프라인 템플릿 탐색

이 섹션에서는 템플릿에 의해 생성된 스캐폴딩에 대한 개요를 제공합니다.

  1. 파이프라인의 루트 디렉터리에 복사된 디렉터리 및 파일 탐색

    • 와 파이프 라인 디렉토리
      • pipeline.py - 파이프 라인을 정의하고 구성 요소를 목록 사용되는
      • configs.py - 같은 데이터가 나있는 오는 지휘자 사용되는 위치로 홀드 구성 세부 사항
    • 데이터 디렉토리
      • 이것은 일반적으로 포함 data.csv 대한 기본 소스 파일, ExampleGen . 당신의 데이터 소스를 변경할 수 있습니다 configs.py .

    • 전처리 코드와 모델 구현을 가진 모델 디렉토리

    • 템플릿은 로컬 환경 및 Kubeflow용 DAG 실행기를 복사합니다.

    • 일부 템플릿에는 기계 학습 메타데이터로 데이터와 아티팩트를 탐색할 수 있도록 Python 노트북도 포함되어 있습니다.

  2. 파이프라인 디렉터리에서 다음 명령을 실행합니다.

    tfx pipeline create --pipeline_path local_runner.py

    tfx run create --pipeline_name pipeline_name

    이 명령은 사용하여 파이프 라인 실행 생성 LocalDagRunner 파이프 라인에 다음 디렉토리를 추가합니다 :

    • 로컬로 사용되는 ML 메타 데이터 저장소를 포함하는 tfx_metadata 디렉토리.
    • 파이프 라인의 파일 출력을 포함하는 tfx_pipeline_output 디렉토리.

  3. 파이프 라인의여 pipeline/configs.py 파일 및 내용을 검토합니다. 이 스크립트는 파이프라인 및 구성 요소 기능에서 사용하는 구성 옵션을 정의합니다. 여기에서 데이터 소스의 위치나 실행의 훈련 단계 수와 같은 항목을 지정할 수 있습니다.

  4. 파이프 라인의여 pipeline/pipeline.py 파일과 내용을 검토합니다. 이 스크립트는 TFX 파이프라인을 생성합니다. 처음에는 파이프 라인 만 포함 ExampleGen 구성 요소를.

    • 에서 TODO 주석의 지시에 따라 pipeline.py 파이프 라인에 더 많은 단계를 추가 할 수 있습니다.

  5. 열기 local_runner.py 파일 및 검토 내용을 표시합니다. 이 스크립트는 파이프 라인 실행을 작성하고 같은과 실행의 매개 변수를 지정 data_path 및 preprocessing_fn .

  6. 당신은 템플릿에 의해 생성 된 비계를 검토하여 파이프 라인 실행 만든 LocalDagRunner . 그런 다음 요구 사항에 맞게 템플릿을 사용자 지정합니다.

파이프라인 사용자 지정

이 섹션에서는 템플릿 사용자 지정을 시작하는 방법에 대한 개요를 제공합니다.

  1. 파이프라인을 설계하십시오. 템플릿이 제공하는 스캐폴딩은 TFX 표준 구성 요소를 사용하여 테이블 형식 데이터에 대한 파이프라인을 구현하는 데 도움이 됩니다. 당신이 파이프 라인에 기존의 ML 워크 플로우를 이동하는 경우, 당신은 최대한 활용하기 위해 코드를 수정해야 할 수도 있습니다 TFX 표준 구성 요소를 . 또한 만들어야 할 수 있습니다 사용자 정의 구성 요소 아직 TFX 표준 구성 요소에서 지원하지 않는 워크 플로우 나에 고유 한 기능을 구현합니다.

  2. 파이프라인을 설계했으면 다음 프로세스를 사용하여 파이프라인을 반복적으로 사용자 지정합니다. 구성 요소에서 시작 일반적으로 당신의 파이프 라인에 섭취 데이터가 ExampleGen 구성 요소입니다.

    1. 사용 사례에 맞게 파이프라인 또는 구성 요소를 사용자 지정합니다. 이러한 사용자 지정에는 다음과 같은 변경 사항이 포함될 수 있습니다.

      • 파이프라인 매개변수 변경
      • 파이프라인에 구성 요소를 추가하거나 제거합니다.
      • 데이터 입력 소스를 교체합니다. 이 데이터 소스는 파일이거나 BigQuery와 같은 서비스에 대한 쿼리일 수 있습니다.
      • 파이프라인에서 구성 요소의 구성을 변경합니다.
      • 구성 요소의 사용자 정의 기능 변경.

    2. 로컬로 사용하여 구성 요소를 실행 local_runner.py 다른 지휘자를 사용하는 경우 스크립트, 또는 다른 적절한 DAG 러너. 스크립트가 실패하면 실패를 디버그하고 스크립트를 다시 실행하십시오.

    3. 이 사용자 정의가 작동하면 다음 사용자 정의로 이동하십시오.

  3. 반복적으로 작업하면 템플릿 워크플로의 각 단계를 필요에 맞게 사용자 지정할 수 있습니다.

사용자 지정 파이프라인 구축

템플릿을 사용하지 않고 사용자 지정 파이프라인을 구축하는 방법에 대해 자세히 알아보려면 다음 지침을 따르십시오.

  1. 파이프라인을 설계하십시오. TFX 표준 구성 요소는 완전한 ML 워크플로를 구현하는 데 도움이 되는 입증된 기능을 제공합니다. 기존 ML 워크플로를 파이프라인으로 이동하는 경우 TFX 표준 구성 요소를 최대한 활용하도록 코드를 수정해야 할 수 있습니다. 또한 만들어야 할 수 있습니다 사용자 정의 구성 요소 등의 데이터 증가 등의 기능을 구현한다.

  2. 다음 예제를 사용하여 파이프라인을 정의하는 스크립트 파일을 생성합니다. 이 가이드는이 파일을 참조 my_pipeline.py .

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

    앞으로 단계에서, 당신은 당신의 파이프 라인을 정의 create_pipeline 로컬 로컬 주자를 사용하여 파이프 라인을 실행합니다.

    다음 프로세스를 사용하여 파이프라인을 반복적으로 빌드합니다.

    1. 사용 사례에 맞게 파이프라인 또는 구성 요소를 사용자 지정합니다. 이러한 사용자 지정에는 다음과 같은 변경 사항이 포함될 수 있습니다.

      • 파이프라인 매개변수 변경
      • 파이프라인에 구성 요소를 추가하거나 제거합니다.
      • 데이터 입력 파일을 교체합니다.
      • 파이프라인에서 구성 요소의 구성을 변경합니다.
      • 구성 요소의 사용자 정의 기능 변경.

    2. 로컬 실행기를 사용하거나 스크립트를 직접 실행하여 구성 요소를 로컬로 실행합니다. 스크립트가 실패하면 실패를 디버그하고 스크립트를 다시 실행하십시오.

    3. 이 사용자 정의가 작동하면 다음 사용자 정의로 이동하십시오.

    파이프라인 워크플로의 첫 번째 노드에서 시작합니다. 일반적으로 첫 번째 노드는 데이터를 파이프라인으로 수집합니다.

  3. 워크플로의 첫 번째 노드를 파이프라인에 추가합니다. 이 예에서, 파이프 라인은 사용 ExampleGen 에서 디렉토리에서 CSV를로드 할 표준 구성 요소를 ./data .

    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 지정된 데이터 경로에 CSV의 데이터를 사용하는 예시적인 레코드를 직렬화 만든다. 설정에 따라 CsvExampleGen 구성 요소의 input_base 데이터 루트로 매개 변수를.

  4. 크리에이트 data 와 동일한 디렉토리에있는 디렉토리 my_pipeline.py . 받는 작은 CSV 파일을 추가 data 디렉토리.

  5. 당신의 실행하려면 다음 명령을 사용하여 my_pipeline.py 스크립트를.

    python my_pipeline.py

    결과는 다음과 같아야 합니다.

    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. 계속해서 반복적으로 파이프라인에 구성 요소를 추가합니다.