Testowanie potoku przy użyciu programów Stub Executors

Wstęp

Aby kontynuować ten samouczek, należy ukończyć samouczek template.ipynb do kroku 6 .

Ten dokument zawiera instrukcje dotyczące testowania potoku TensorFlow Extended (TFX) przy użyciu BaseStubExecuctor , który generuje fałszywe artefakty przy użyciu złotych danych testowych. Jest to przeznaczone dla użytkowników, którzy mogą zastąpić executory, których nie chcą testować, aby zaoszczędzić czas na uruchamianiu rzeczywistych executorów. Executor kodu pośredniczącego jest dostarczany z pakietem TFX Python pod adresem tfx.experimental.pipeline_testing.base_stub_executor .

Ten samouczek stanowi rozszerzenie samouczka template.ipynb , dlatego będziesz korzystać także ze zbioru danych Taxi Trips wydanego przez miasto Chicago. Gorąco zachęcamy do wypróbowania modyfikacji komponentów przed użyciem modułów wykonawczych pośredniczących.

1. Zapisz dane wyjściowe potoku w Google Cloud Storage

Najpierw musimy zarejestrować dane wyjściowe potoku, aby wykonawcy kodu pośredniczącego mogli skopiować artefakty z zarejestrowanych wyników.

Ponieważ w tym samouczku założono, że template.ipynb został ukończony do kroku 6, pomyślne uruchomienie potoku musi zostać zapisane w pliku MLMD . Dostęp do informacji o wykonaniu w MLMD można uzyskać za pomocą serwera gRPC.

Otwórz terminal i uruchom następujące polecenia:

  1. Wygeneruj plik kubeconfig z odpowiednimi poświadczeniami: bash gcloud container clusters get-credentials $cluster_name --zone $compute_zone --project $gcp_project_id $compute_zone to region dla silnika Gcp, a $gcp_project_id to identyfikator projektu Twojego projektu GCP.

  2. Skonfiguruj przekierowanie portów w celu połączenia z MLMD: bash nohup kubectl port-forward deployment/metadata-grpc-deployment -n $namespace $port:8080 & $namespace to przestrzeń nazw klastra, a $port to dowolny nieużywany port, który będzie używany przekierowanie portów.

  3. Sklonuj repozytorium tfx GitHub. W katalogu tfx uruchom następującą komendę: 

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 powinna być ustawiona na ścieżkę w Google Cloud Storage, w której mają być rejestrowane dane wyjściowe potoku, dlatego pamiętaj o zastąpieniu <gcp_project_id> identyfikatorem projektu GCP.

$host i $port to nazwa hosta i port serwera grpc metadanych służącego do połączenia z MLMD. $port powinien być ustawiony na numer portu, którego użyłeś do przekierowania portów, a jako nazwę hosta możesz ustawić „localhost”.

W samouczku template.ipynb nazwa potoku jest domyślnie ustawiona jako „my_pipeline”, więc ustaw pipeline_name="my_pipeline" . Jeśli zmodyfikowałeś nazwę potoku podczas uruchamiania samouczka szablonu, powinieneś odpowiednio zmodyfikować --pipeline_name .

2. Włącz executory Stub w Kubeflow DAG Runner

Najpierw upewnij się, że predefiniowany szablon został skopiowany do katalogu projektu za pomocą polecenia CLI tfx template copy . W skopiowanych plikach źródłowych konieczna jest edycja dwóch następujących plików.

  1. Utwórz plik o nazwie stub_component_launcher.py w katalogu, w którym znajduje się kubeflow_dag_runner.py i umieść w nim następującą zawartość.

    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. Ustaw identyfikatory komponentów na listę identyfikatorów komponentów, które mają zostać przetestowane (innymi słowy, executory innych komponentów zostaną zastąpione przez BaseStubExecutor).

  3. Otwórz kubeflow_dag_runner.py . Dodaj następującą instrukcję importu na górze, aby użyć właśnie dodanej klasy StubComponentLauncher .

    import stub_component_launcher

  4. W kubeflow_dag_runner.py dodaj klasę StubComponentLauncher do klasy supported_launcher_class KubeflowDagRunnerConfig , aby umożliwić uruchamianie modułów wykonawczych kodu pośredniczącego:

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

3. Zaktualizuj i uruchom potok za pomocą modułów wykonawczych

Zaktualizuj istniejący potok za pomocą zmodyfikowanej definicji potoku za pomocą modułów wykonawczych.

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

$endpoint powinien być ustawiony na punkt końcowy klastra KFP.

Uruchom następujące polecenie, aby utworzyć nowe uruchomienie wykonania zaktualizowanego potoku.

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

Sprzątanie

Użyj polecenia fg , aby uzyskać dostęp do przekierowania portów w tle, a następnie Ctrl-C, aby zakończyć. Możesz usunąć katalog z zarejestrowanymi wynikami potoku, używając gsutil -m rm -R $output_dir .

Aby wyczyścić wszystkie zasoby Google Cloud użyte w tym projekcie, możesz usunąć projekt Google Cloud użyty w samouczku.

Alternatywnie możesz wyczyścić poszczególne zasoby, odwiedzając każdą konsolę: - Google Cloud Storage - Google Container Registry - Google Kubernetes Engine