Testando o pipeline usando Stub Executors

Introdução

Você deve concluir o tutorial template.ipynb até a Etapa 6 para prosseguir com este tutorial.

Este documento fornecerá instruções para testar um pipeline TensorFlow Extended (TFX) usando BaseStubExecuctor , que gera artefatos falsos usando os dados de teste dourados. O objetivo é que os usuários substituam os executores que não desejam testar, para que possam economizar tempo na execução dos executores reais. O executor stub é fornecido com o pacote TFX Python em tfx.experimental.pipeline_testing.base_stub_executor .

Este tutorial serve como uma extensão do tutorial template.ipynb , portanto, você também usará o conjunto de dados Taxi Trips lançado pela cidade de Chicago. Recomendamos fortemente que você tente modificar os componentes antes de utilizar executores stub.

1. Registre as saídas do pipeline no Google Cloud Storage

Primeiro precisamos registrar as saídas do pipeline para que os executores de stub possam copiar os artefatos das saídas registradas.

Como este tutorial pressupõe que você concluiu template.ipynb até a etapa 6, uma execução bem-sucedida do pipeline deve ter sido salva no MLMD . As informações de execução no MLMD podem ser acessadas usando o servidor gRPC.

Abra um Terminal e execute os seguintes comandos:

1. Gere um arquivo kubeconfig com credenciais apropriadas: bash gcloud container clusters get-credentials $cluster_name --zone $compute_zone --project $gcp_project_id $compute_zone é a região do mecanismo gcp e $gcp_project_id é o ID do projeto do seu projeto GCP.

2. Configure o encaminhamento de porta para conexão com o MLMD: bash nohup kubectl port-forward deployment/metadata-grpc-deployment -n $namespace $port:8080 & $namespace é o namespace do cluster e $port é qualquer porta não utilizada que será usada para encaminhamento de porta.

3. Clone o repositório tfx GitHub. Dentro do diretório tfx, execute o seguinte comando:

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 deve ser definido como um caminho no Google Cloud Storage onde as saídas do pipeline serão registradas. Portanto, substitua <gcp_project_id> pelo ID do projeto do GCP.

$host e $port são o nome do host e a porta do servidor grpc de metadados para conectar-se ao MLMD. $port deve ser definido como o número da porta que você usou para encaminhamento de porta e você pode definir "localhost" para o nome do host.

No tutorial template.ipynb , o nome do pipeline é definido como "my_pipeline" por padrão, então defina pipeline_name="my_pipeline" . Se você modificou o nome do pipeline ao executar o tutorial do modelo, deverá modificar --pipeline_name adequadamente.

2. Habilite executores de stub no Kubeflow DAG Runner

Primeiro, certifique-se de que o modelo predefinido foi copiado para o diretório do projeto usando o comando CLI tfx template copy . É necessário editar os dois arquivos a seguir nos arquivos de origem copiados.

1. Crie um arquivo chamado stub_component_launcher.py no diretório onde kubeflow_dag_runner.py está localizado e coloque o seguinte conteúdo nele.

   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. Defina os IDs dos componentes como uma lista de IDs dos componentes que serão testados (em outras palavras, os executores de outros componentes são substituídos por BaseStubExecutor).

3. Abra kubeflow_dag_runner.py . Adicione a seguinte instrução de importação na parte superior para usar a classe StubComponentLauncher que acabamos de adicionar.

   import stub_component_launcher
   

4. Em kubeflow_dag_runner.py , adicione a classe StubComponentLauncher à supported_launcher_class de KubeflowDagRunnerConfig para permitir o lançamento de executores de stub:

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

3. Atualize e execute o pipeline com executores de stub

Atualize o pipeline existente com a definição de pipeline modificada com executores de stub.

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

$endpoint deve ser definido como o endpoint do cluster KFP.

Execute o comando a seguir para criar uma nova execução do pipeline atualizado.

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

Limpando

Use o comando fg para acessar o encaminhamento de porta em segundo plano e depois ctrl-C para encerrar. Você pode excluir o diretório com saídas de pipeline gravadas usando gsutil -m rm -R $output_dir .

Para limpar todos os recursos do Google Cloud usados ​​neste projeto, exclua o projeto do Google Cloud usado no tutorial.

Alternativamente, você pode limpar recursos individuais visitando cada console: - Google Cloud Storage - Google Container Registry - Google Kubernetes Engine