Kiểm tra đường ống bằng cách sử dụng Stub Executor

Giới thiệu

Bạn nên hoàn thành hướng dẫn template.ipynb cho đến Bước 6 để tiếp tục hướng dẫn này.

Tài liệu này sẽ cung cấp hướng dẫn để kiểm tra đường dẫn TensorFlow Extended (TFX) bằng cách sử dụng BaseStubExecuctor , tạo ra các thành phần giả bằng cách sử dụng dữ liệu thử nghiệm vàng. Điều này nhằm mục đích giúp người dùng thay thế các trình thực thi mà họ không muốn kiểm tra để họ có thể tiết kiệm thời gian chạy các trình thực thi thực tế. Trình thực thi sơ khai được cung cấp gói TFX Python trong tfx.experimental.pipeline_testing.base_stub_executor .

Hướng dẫn này đóng vai trò như một phần mở rộng cho hướng dẫn template.ipynb , do đó bạn cũng sẽ sử dụng tập dữ liệu Taxi Trips do Thành phố Chicago phát hành. Chúng tôi đặc biệt khuyến khích bạn thử sửa đổi các thành phần trước khi sử dụng trình thực thi sơ khai.

1. Ghi lại kết quả đầu ra của quy trình trong Google Cloud Storage

Trước tiên, chúng tôi cần ghi lại các đầu ra của quy trình để người thực thi sơ khai có thể sao chép các tạo phẩm từ các đầu ra đã ghi.

Vì hướng dẫn này giả định rằng bạn đã hoàn thành template.ipynb cho đến bước 6 nên quá trình chạy quy trình thành công phải được lưu trong MLMD . Thông tin thực thi trong MLMD có thể được truy cập bằng máy chủ gRPC.

Mở Terminal và chạy các lệnh sau:

1. Tạo tệp kubeconfig với thông tin xác thực phù hợp: bash gcloud container clusters get-credentials $cluster_name --zone $compute_zone --project $gcp_project_id $compute_zone là vùng dành cho công cụ gcp và $gcp_project_id là id dự án của dự án GCP của bạn.

2. Thiết lập chuyển tiếp cổng để kết nối với MLMD: bash nohup kubectl port-forward deployment/metadata-grpc-deployment -n $namespace $port:8080 & $namespace là không gian tên cụm và $port là bất kỳ cổng nào không được sử dụng sẽ được sử dụng cho chuyển tiếp cổng.

3. Sao chép kho lưu trữ tfx GitHub. Trong thư mục tfx, chạy lệnh sau:

python tfx/experimental/pipeline_testing/pipeline_recorder.py \
--output_dir=gs://<gcp_project_id>-kubeflowpipelines-default/testdata \
--host=$host \
--port=$port \
--pipeline_name=$pipeline_name

$output_dir phải được đặt thành một đường dẫn trong Google Cloud Storage nơi ghi lại các đầu ra của đường dẫn, vì vậy hãy đảm bảo thay thế <gcp_project_id> bằng id dự án GCP.

$host và $port là tên máy chủ và cổng của máy chủ grpc siêu dữ liệu để kết nối với MLMD. $port phải được đặt thành số cổng bạn đã sử dụng để chuyển tiếp cổng và bạn có thể đặt "localhost" cho tên máy chủ.

Trong hướng dẫn template.ipynb , tên đường dẫn được đặt là "my_pipeline" theo mặc định, vì vậy hãy đặt pipeline_name="my_pipeline" . Nếu bạn đã sửa đổi tên đường dẫn khi chạy hướng dẫn mẫu, bạn nên sửa đổi --pipeline_name cho phù hợp.

2. Kích hoạt Stub Executor trong Kubeflow DAG Runner

Trước tiên, hãy đảm bảo rằng mẫu được xác định trước đã được sao chép vào thư mục dự án của bạn bằng lệnh CLI tfx template copy . Cần phải chỉnh sửa hai tệp sau trong tệp nguồn được sao chép.

1. Tạo một tệp có tên stub_component_launcher.py trong thư mục chứa kubeflow_dag_runner.py và đặt nội dung sau vào đó.

   from tfx.experimental.pipeline_testing import base_stub_component_launcher
   from pipeline import configs
   
   class StubComponentLauncher(
       base_stub_component_launcher.BaseStubComponentLauncher):
     pass
   
   # GCS directory where KFP outputs are recorded
   test_data_dir = "gs://{}/testdata".format(configs.GCS_BUCKET_NAME)
   # TODO: customize self.test_component_ids to test components, replacing other
   # component executors with a BaseStubExecutor.
   test_component_ids = ['Trainer']
   StubComponentLauncher.initialize(
       test_data_dir=test_data_dir,
       test_component_ids=test_component_ids)
   

2. Đặt id thành phần làm danh sách id thành phần sẽ được kiểm tra (nói cách khác, bộ thực thi của các thành phần khác được thay thế bằng BaseStubExecutor).

3. Mở kubeflow_dag_runner.py . Thêm câu lệnh nhập sau vào đầu để sử dụng lớp StubComponentLauncher mà chúng tôi vừa thêm.

   import stub_component_launcher
   

4. Trong kubeflow_dag_runner.py , thêm lớp StubComponentLauncher vào supported_launcher_class của KubeflowDagRunnerConfig để cho phép khởi chạy các trình thực thi sơ khai:

   runner_config = kubeflow_dag_runner.KubeflowDagRunnerConfig(
       supported_launcher_classes=[
           stub_component_launcher.StubComponentLauncher
       ],
   

3. Cập nhật và chạy quy trình với trình thực thi sơ khai

Cập nhật quy trình hiện có với định nghĩa quy trình đã sửa đổi với trình thực thi sơ khai.

tfx pipeline update --pipeline-path=kubeflow_dag_runner.py \
  --endpoint=$endpoint --engine=kubeflow

$endpoint phải được đặt thành điểm cuối cụm KFP của bạn.

Chạy lệnh sau để tạo một lần chạy thực thi mới cho quy trình đã cập nhật của bạn.

tfx run create --pipeline-name $pipeline_name --endpoint=$endpoint \
  --engine=kubeflow

Dọn dẹp

Sử dụng lệnh fg để truy cập chuyển tiếp cổng ở chế độ nền, sau đó ctrl-C để kết thúc. Bạn có thể xóa thư mục chứa các đầu ra đường dẫn đã ghi bằng cách sử dụng gsutil -m rm -R $output_dir .

Để dọn sạch tất cả tài nguyên Google Cloud được sử dụng trong dự án này, bạn có thể xóa dự án Google Cloud mà bạn đã sử dụng cho hướng dẫn.

Ngoài ra, bạn có thể dọn sạch các tài nguyên riêng lẻ bằng cách truy cập từng bảng điều khiển: - Google Cloud Storage - Google Container Register - Google Kubernetes Engine