Xây dựng một đường ống TFX cục bộ

TFX giúp dễ dàng sắp xếp quy trình công việc học máy (ML) của bạn dưới dạng một đường dẫn, để:

  • Tự động hóa quy trình ML của bạn, cho phép bạn thường xuyên đào tạo lại, đánh giá và triển khai mô hình của mình.
  • Tạo đường ống ML bao gồm phân tích sâu về hiệu suất của mô hình và xác nhận các mô hình mới được đào tạo để đảm bảo hiệu suất và độ tin cậy.
  • Theo dõi dữ liệu đào tạo để tìm các điểm bất thường và loại bỏ sai lệch phục vụ đào tạo
  • Tăng tốc độ thử nghiệm bằng cách chạy một đường ống với các bộ siêu tham số khác nhau.

Quá trình phát triển đường ống điển hình bắt đầu trên một máy cục bộ, với phân tích dữ liệu và thiết lập thành phần, trước khi được triển khai vào sản xuất. Hướng dẫn này mô tả hai cách để xây dựng một đường ống cục bộ.

  • Tùy chỉnh mẫu đường dẫn TFX để phù hợp với nhu cầu của quy trình làm việc ML của bạn. Các mẫu đường ống TFX là các quy trình làm việc được tạo sẵn nhằm thể hiện các phương pháp hay nhất bằng cách sử dụng các thành phần tiêu chuẩn TFX.
  • Xây dựng đường dẫn bằng TFX. Trong trường hợp sử dụng này, bạn xác định một đường dẫn mà không bắt đầu từ một mẫu.

Khi bạn đang phát triển đường ống của bạn, bạn có thể chạy nó với LocalDagRunner . Sau đó, khi các thành phần của đường ống đã được xác định và kiểm tra rõ ràng, bạn sẽ sử dụng bộ điều phối cấp sản xuất như Kubeflow hoặc Airflow.

Trước khi bắt đầu

TFX là một gói Python, vì vậy bạn sẽ cần thiết lập môi trường phát triển Python, chẳng hạn như môi trường ảo hoặc vùng chứa Docker. Sau đó:

pip install tfx

Nếu bạn chưa quen với đường ống TFX, tìm hiểu thêm về những khái niệm cốt lõi cho đường ống TFX trước khi tiếp tục.

Xây dựng đường dẫn bằng cách sử dụng mẫu

Mẫu đường ống TFX giúp bạn bắt đầu phát triển đường ống dễ dàng hơn bằng cách cung cấp một bộ định nghĩa đường ống được tạo sẵn mà bạn có thể tùy chỉnh cho trường hợp sử dụng của mình.

Các phần sau đây mô tả cách tạo bản sao của một mẫu và tùy chỉnh nó để đáp ứng nhu cầu của bạn.

Tạo một bản sao của mẫu đường ống

  1. Xem danh sách các mẫu đường dẫn TFX có sẵn:

    tfx template list

  2. Chọn một mẫu từ danh sách

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

    Thay thế những điều sau:

    • template : Tên của mẫu bạn muốn sao chép.
    • pipeline-name : Tên của các đường ống dẫn để tạo ra.
    • destination-path : Đường dẫn để sao chép các mẫu vào.

    Tìm hiểu thêm về các tfx template copy lệnh .

  3. Một bản sao của mẫu đường ống đã được tạo tại đường dẫn bạn đã chỉ định.

Khám phá mẫu đường ống

Phần này cung cấp một cái nhìn tổng quan về giàn giáo được tạo bởi một mẫu.

  1. Khám phá các thư mục và tệp đã được sao chép vào thư mục gốc của đường dẫn của bạn

    • Một thư mục đường ống với
      • pipeline.py - xác định đường ống, và danh sách các thành phần đang được sử dụng
      • configs.py chi tiết cấu hình giữ như nơi dữ liệu đến từ hoặc có Orchestrator đang được sử dụng -
    • Một thư mục dữ liệu
      • Điều này thường chứa một data.csv tập tin, đó là nguồn gốc mặc định cho ExampleGen . Bạn có thể thay đổi nguồn dữ liệu trong configs.py .

    • Một mô hình thư mục với tiền xử lý mã và mô hình triển khai

    • Mẫu sao chép các trình chạy DAG cho môi trường cục bộ và Kubeflow.

    • Một số mẫu cũng bao gồm Sổ ghi chép Python để bạn có thể khám phá dữ liệu và hiện vật của mình bằng Siêu dữ liệu học máy.

  2. Chạy các lệnh sau trong thư mục đường dẫn của bạn:

    tfx pipeline create --pipeline_path local_runner.py

    tfx run create --pipeline_name pipeline_name

    Lệnh tạo ra một đường ống chạy bằng LocalDagRunner , có thêm các thư mục sau đây để đường ống của bạn:

    • Một thư mục tfx_metadata trong đó có các cửa hàng ML Metadata sử dụng tại địa phương.
    • Một thư mục tfx_pipeline_output trong đó có kết quả đầu ra hồ sơ của đường ống.

  3. Mở của đường ống của bạn pipeline/configs.py file và xem lại nội dung. Tập lệnh này xác định các tùy chọn cấu hình được sử dụng bởi đường ống và các chức năng thành phần. Đây là nơi bạn sẽ chỉ định những thứ như vị trí của nguồn dữ liệu hoặc số bước đào tạo trong một lần chạy.

  4. Mở của đường ống của bạn pipeline/pipeline.py file và xem lại nội dung. Tập lệnh này tạo đường dẫn TFX. Ban đầu, đường ống chỉ chứa một ExampleGen thành phần.

    • Làm theo hướng dẫn trong các ý kiến TODO trong pipeline.py để bổ sung thêm các bước để các đường ống dẫn.

  5. Mở local_runner.py tập tin và rà soát các nội dung. Kịch bản này tạo ra một hoạt động đường ống dẫn và quy định cụ thể các thông số của chạy, chẳng hạn như data_path và preprocessing_fn .

  6. Bạn đã xem xét các giàn giáo được tạo ra bởi mẫu và tạo ra một đường ống chạy bằng LocalDagRunner . Tiếp theo, tùy chỉnh mẫu để phù hợp với yêu cầu của bạn.

Tùy chỉnh đường dẫn của bạn

Phần này cung cấp tổng quan về cách bắt đầu tùy chỉnh mẫu của bạn.

  1. Thiết kế đường ống của bạn. Khung mà một mẫu cung cấp giúp bạn triển khai một đường dẫn cho dữ liệu dạng bảng bằng cách sử dụng các thành phần tiêu chuẩn TFX. Nếu bạn đang di chuyển một workflow ML hiện có vào một đường ống dẫn, bạn có thể cần phải sửa đổi mã của bạn để tận dụng tối đa các thành phần tiêu chuẩn TFX . Bạn cũng có thể cần phải tạo các thành phần tùy chỉnh mà thực hiện tính năng mà là duy nhất cho công việc của bạn hoặc chưa được hỗ trợ bởi các thành phần tiêu chuẩn TFX.

  2. Khi bạn đã thiết kế xong đường dẫn của mình, hãy tùy chỉnh đường dẫn lặp đi lặp lại bằng quy trình sau. Bắt đầu từ thành phần ingests dữ liệu vào đường ống của bạn, mà thường là ExampleGen thành phần.

    1. Tùy chỉnh đường ống hoặc một thành phần để phù hợp với trường hợp sử dụng của bạn. Những tùy chỉnh này có thể bao gồm những thay đổi như:

      • Thay đổi thông số đường ống.
      • Thêm các thành phần vào đường ống hoặc loại bỏ chúng.
      • Thay thế nguồn đầu vào dữ liệu. Nguồn dữ liệu này có thể là một tệp hoặc các truy vấn vào các dịch vụ như BigQuery.
      • Thay đổi cấu hình của một thành phần trong đường dẫn.
      • Thay đổi chức năng tùy chỉnh của một thành phần.

    2. Chạy phần tại địa phương bằng cách sử dụng local_runner.py kịch bản, hoặc một Á hậu DAG thích hợp nếu bạn đang sử dụng một Orchestrator khác nhau. Nếu tập lệnh không thành công, hãy gỡ lỗi và thử chạy lại tập lệnh.

    3. Khi tùy chỉnh này hoạt động, hãy chuyển sang tùy chỉnh tiếp theo.

  3. Làm việc lặp đi lặp lại, bạn có thể tùy chỉnh từng bước trong quy trình làm việc mẫu để đáp ứng nhu cầu của mình.

Xây dựng một đường ống tùy chỉnh

Sử dụng các hướng dẫn sau để tìm hiểu thêm về cách xây dựng đường ống tùy chỉnh mà không cần sử dụng mẫu.

  1. Thiết kế đường ống của bạn. Các thành phần tiêu chuẩn TFX cung cấp chức năng đã được chứng minh để giúp bạn triển khai quy trình làm việc ML hoàn chỉnh. Nếu bạn đang di chuyển một quy trình ML hiện có vào một đường ống, bạn có thể cần phải sửa đổi mã của mình để sử dụng đầy đủ các thành phần tiêu chuẩn TFX. Bạn cũng có thể cần phải tạo các thành phần tùy chỉnh mà thực hiện các chức năng như dữ liệu tăng thêm.

  2. Tạo một tệp kịch bản để xác định đường dẫn của bạn bằng cách sử dụng ví dụ sau. Hướng dẫn này đề cập đến tập tin này như 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()

    Trong các bước tiếp theo, bạn xác định đường ống của bạn trong create_pipeline và chạy đường ống của bạn tại địa phương bằng cách sử dụng Á hậu địa phương.

    Xây dựng lặp đi lặp lại đường dẫn của bạn bằng quy trình sau.

    1. Tùy chỉnh đường ống hoặc một thành phần để phù hợp với trường hợp sử dụng của bạn. Những tùy chỉnh này có thể bao gồm những thay đổi như:

      • Thay đổi thông số đường ống.
      • Thêm các thành phần vào đường ống hoặc loại bỏ chúng.
      • Thay thế tệp đầu vào dữ liệu.
      • Thay đổi cấu hình của một thành phần trong đường dẫn.
      • Thay đổi chức năng tùy chỉnh của một thành phần.

    2. Chạy cục bộ thành phần bằng trình chạy cục bộ hoặc bằng cách chạy trực tiếp tập lệnh. Nếu tập lệnh không thành công, hãy gỡ lỗi và thử chạy lại tập lệnh.

    3. Khi tùy chỉnh này hoạt động, hãy chuyển sang tùy chỉnh tiếp theo.

    Bắt đầu từ nút đầu tiên trong quy trình làm việc của đường dẫn của bạn, thường là nút đầu tiên nhập dữ liệu vào đường dẫn của bạn.

  3. Thêm nút đầu tiên trong quy trình làm việc vào đường dẫn của bạn. Trong ví dụ này, các đường ống dẫn sử dụng ExampleGen thành phần tiêu chuẩn để tải một CSV từ một thư mục tại ./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 sẽ tạo các đăng dụ hồ sơ bằng cách sử dụng dữ liệu trong CSV tại đường dẫn dữ liệu cụ thể. Bằng cách đặt các CsvExampleGen thành phần của input_base tham số với gốc dữ liệu.

  4. Tạo một data thư mục trong thư mục tương tự như my_pipeline.py . Thêm một tập tin CSV nhỏ để các data thư mục.

  5. Sử dụng lệnh sau để chạy bạn my_pipeline.py kịch bản.

    python my_pipeline.py

    Kết quả sẽ giống như sau:

    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. Tiếp tục thêm lặp đi lặp lại các thành phần vào đường dẫn của bạn.