> For the complete documentation index, see [llms.txt](https://aurimrv.gitbook.io/pratica-devops-com-docker-para-machine-learning/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://aurimrv.gitbook.io/pratica-devops-com-docker-para-machine-learning/id-11-mlops-com-kubeflow/11.3-kubeflow-pipeline.md).

# 11.3 Kubeflow Pipeline

O [Kubeflow Pipeline](https://www.kubeflow.org/docs/components/pipelines/overview/) é uma plataforma para workflows de Machine Learning baseada em contêineres. Os pipelines são desenvolvidos com a linguagem Python e traduzidos em objetos dentro do Kubernetes. Revise os [conceitos](https://www.kubeflow.org/docs/components/pipelines/concepts/) do Pipeline para melhor entendimento.

## 11.3.1 O que há de novo no KFP v2?

Com o lançamento do Kubeflow v1.11.0, o **Kubeflow Pipelines v2 (KFP v2)** é o padrão consolidado. A versão 2 traz mudanças arquiteturais significativas em relação à versão v1:

* **Especificação Baseada em IR (Intermediate Representation):** Ao invés de compilar pipelines em arquivos `.zip` ou `.tar.gz` contendo definições específicas do Argo Workflows (Kubernetes), o KFP v2 compila os pipelines em um arquivo **YAML de representação intermediária (IR)**. Essa representação é independente da engine de execução subjacente.
* **Decoração Explícita de Componentes:** O uso do decorador `@dsl.component` define componentes baseados em funções Python, enquanto `@dsl.container_component` substitui as antigas abordagens para componentes baseados em imagens Docker customizadas de maneira muito mais robusta.
* **Tipagem Forte de Dados (Artifacts):** O KFP v2 gerencia a linhagem de dados usando objetos `Input[Artifact]` e `Output[Artifact]` (como `Dataset`, `Model`, `Metrics`). Isso permite que os metadados de execução sejam rastreados de forma integrada e visualizados na WebUI do Kubeflow.
* **Integração Nativa com o Model Registry:** Facilita o registro automático de modelos treinados em pipelines diretamente no repositório de modelos da plataforma.

A seguir, vamos construir um exemplo prático utilizando o SDK do KFP v2.

## 11.3.2 Configuração do Notebook

Antes de iniciar o exemplo, precisamos criar uma [configuração](https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/connect-api/) `PodDefault` para que as execuções do SDK do KFP tenham a permissão necessária:

```bash
cat <<EOF | kubectl apply -f -
apiVersion: kubeflow.org/v1alpha1
kind: PodDefault
metadata:
  name: access-ml-pipeline
  namespace: "kubeflow-user-example-com"
spec:
  desc: Allow access to Kubeflow Pipelines
  selector:
    matchLabels:
      access-ml-pipeline: "true"
  volumes:
    - name: volume-kf-pipeline-token
      projected:
        sources:
          - serviceAccountToken:
              path: token
              expirationSeconds: 7200
              audience: pipelines.kubeflow.org
  volumeMounts:
    - mountPath: /var/run/secrets/kubeflow/pipelines
      name: volume-kf-pipeline-token
      readOnly: true
  env:
    - name: KF_PIPELINES_SA_TOKEN_PATH
      value: /var/run/secrets/kubeflow/pipelines/token
EOF
```

Execute o comando acima para criar o recurso diretamente no Kubernetes. Agora podemos acessar o Notebook do Kubeflow e usar a configuração para acessar o Kubeflow Pipeline.

> \[!IMPORTANT] **Atenção ao criar o Notebook na WebUI:** Durante o processo de criação do Notebook Jupyter através da interface gráfica (WebUI) do Kubeflow, na seção **Configurations**, você **deve marcar a opção/checkbox** correspondente ao `access-ml-pipeline` (com a descrição `Allow access to Kubeflow Pipelines`).
>
> Se você não habilitar essa configuração, o pod do Notebook não receberá o token de autenticação montado no diretório `/var/run/secrets/kubeflow/pipelines/token`. Isso causará erros de `ApiException: (401) Unauthorized` ao tentar executar códigos com o SDK do KFP (`Client`). Caso já tenha criado o Notebook sem marcar essa opção, exclua-o e crie um novo ativando essa opção.

Através da WebUI do Kubeflow, vamos criar um Notebook Jupyter e executar o primeiro pipeline.

Vamos usar o namespace `kubeflow-user-example-com` (criado na configuração do Kubeflow) e instalar as bibliotecas Python para manipular o pipeline via Notebook. Utilize a [documentação oficial](https://www.kubeflow.org/docs/components/notebooks/quickstart-guide/) para criar o Notebook Jupyter.

No Notebook Jupyter, execute a seguinte linha para criar uma variável a ser utilizada no namespace correto:

```python
namespace="kubeflow-user-example-com"
```

## 11.3.3 Primeiros Pipelines

Agora vamos criar o primeiro pipeline:

```python
from kfp import dsl
from kfp import compiler
from kfp.client import Client

@dsl.component
def say_hello(name: str) -> str:
    hello_text = f'Hello, {name}!'
    print(hello_text)
    return hello_text

@dsl.pipeline
def hello_pipeline(recipient: str) -> str:
    hello_task = say_hello(name=recipient)
    return hello_task.output

compiler.Compiler().compile(hello_pipeline, 'pipeline.yaml')

client = Client()
run = client.create_run_from_pipeline_package(
    'pipeline.yaml',
    arguments={
        'recipient': 'World',
    },
    experiment_name='Ola Mundo Experimento',
    run_name='execucao-inicial-teste'
)

```

Verifique na WebUI do Kubeflow a execução do pipeline. Repare que utilizamos as funções do SDK para compilar e criar o pipeline.

### Entendendo a Organização do Kubeflow: Pipelines, Experiments e Runs

Ao navegar pela interface do Kubeflow Pipelines, você notará guias separadas para **Pipelines**, **Experiments** e **Runs**. É importante entender a diferença entre elas:

* **Pipeline**: É o modelo estático (blueprint) do seu fluxo de trabalho de ML. Ele define a estrutura de componentes (grafo DAG), dependências, entradas e saídas.
* **Experiment**: É um espaço de trabalho ou agrupador lógico (como uma pasta) usado para organizar e comparar execuções relacionadas (ex: comparar diferentes execuções de treinamento de um mesmo modelo).
* **Run**: É uma execução de fato de um Pipeline com parâmetros específicos.

> \[!NOTE] **Por que a aba "Pipelines" está vazia após a execução do script acima?**
>
> Ao chamar a função `client.create_run_from_pipeline_package()`, o SDK cria uma execução (**Run**) diretamente a partir do arquivo compactado local (`pipeline.yaml`). O Kubeflow executa os pods, cria o grafo na seção de **Runs** associado a um experimento, mas **não registra** o modelo do pipeline no catálogo global.
>
> Caso deseje registrar o template do pipeline permanentemente no catálogo para visualizá-lo na aba **Pipelines**, você pode fazer o upload manualmente pela interface ou usar a função `client.upload_pipeline()` no Python SDK:
>
> ```python
> client.upload_pipeline(
>     pipeline_package_path='pipeline.yaml',
>     pipeline_name='meu-primeiro-pipeline'
> )
> ```

Agora vamos criar um exemplo de steps (passos) sequenciais. Neste exemplo, ele baixa um arquivo TXT do Cloud Storage e, em seguida, faz o print do resultado. Para que os steps sejam sequenciais, é necessário que a saída de um step seja a entrada do seguinte, ou podemos declarar a dependência com a função `after`:

```python
@dsl.container_component
def gcs_download_op(url: str):
    return dsl.ContainerSpec(
        image='google/cloud-sdk:279.0.0',
        command=['sh', '-c'],
        args=['gsutil cat $0 | tee $1', url, '/tmp/results.txt']
    )
@dsl.container_component
def echo_op(text: str):
    return dsl.ContainerSpec(
        image='library/bash:4.4.23',
        command=['sh', '-c'],
        args=['echo "$0"', text]
    )

@dsl.pipeline(
    name='sequential-pipeline',
    description='A pipeline with two sequential steps.'
)
def sequential_pipeline(url : str='gs://ml-pipeline/sample-data/shakespeare/shakespeare1.txt'):
    """A pipeline with two sequential steps."""
    download_task = gcs_download_op(url=url)
    echo_task = echo_op(text='/tmp/results.txt').after(download_task)
    echo_task2 = echo_op(text='/tmp/results.txt').after(echo_task)

compiler.Compiler().compile(sequential_pipeline, 'sequential_pipeline.yaml')

client = Client()
run = client.create_run_from_pipeline_package(
    'sequential_pipeline.yaml',
    experiment_name='Sequential Pipeline Experimento',
    run_name='execucao-sequencial-1'
)

```

Após a execução deste pipeline, veja a execução na WebUI do Kubeflow.

Existem muitos [exemplos](https://github.com/kubeflow/pipelines/tree/master/samples) de pipeline no repositório do Kubeflow. Teste outros exemplos, como o [Data passing in python components](https://github.com/kubeflow/pipelines/blob/master/samples/tutorials/Data%20passing%20in%20python%20components/Data%20passing%20in%20python%20components%20-%20Files.py).