Criar pipelines e implantá-los¶

Preparar código¶

Antes de operacionalizar seu pipeline, prepare seu código para produção. Se você começou com notebooks, comece reestruturando seu código em funções modulares e reutilizáveis, nas quais cada etapa principal (preparação de dados, engenharia de recursos, treinamento de modelos, avaliação) se torna uma função separada com entradas e saídas claras. Se você já tem scripts modulares, certifique-se de que cada função tenha interfaces e responsabilidades bem definidas. Parametrize todos os valores de configuração, como nomes de tabelas e hiperparâmetros, para permitir a implantação entre ambientes. Recomendamos também criar um script de ponto de entrada que execute o pipeline de ponta a ponta localmente para depuração e desenvolvimento futuro.

Exemplo de estrutura de diretório:

ml_pipeline_project/
├── README.md
├── requirements.txt
├── config/
├── src/ml_pipeline/
│   ├── utils/                     # Common utilities
│   ├── data/                      # Data preparation
│   ├── features/                  # Feature engineering
│   ├── models/                    # Model training
│   └── inference/                 # Model inference
├── scripts/
│   ├── run_pipeline.py            # Main entry point
│   └── dag.py
├── tests/
└── notebooks/

Exemplo de script run_pipeline.py:

import argparse
from ml_pipeline.utils.config_loader import load_config
from ml_pipeline.data.ingestion import load_raw_data
from ml_pipeline.data.validation import validate_data_quality
from ml_pipeline.features.transformers import create_features
from ml_pipeline.models.training import train_model
from ml_pipeline.models.evaluation import evaluate_model
from ml_pipeline.models.registry import register_model

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--config", required=True, help="Config file path")
    parser.add_argument("--env", default="dev", help="Environment (dev/prod)")
    args = parser.parse_args()

    # Load configuration
    config = load_config(args.config, args.env)

    # Execute pipeline stages
    raw_data = load_raw_data(config.data.source_table)
    validate_data_quality(raw_data, config.data.quality_checks)
    features = create_features(raw_data, config.features.transformations)
    model = train_model(features, config.model.hyperparameters)
    metrics = evaluate_model(model, features, config.model.eval_metrics)
    register_model(model, metrics, config.model.registry_name)

if __name__ == "__main__":
    main()

Migração de notebooks para trabalhos de ML¶

A maioria dos códigos escritos nos notebooks Snowflake funcionará em trabalhos e ML sem alterações de código necessárias. Os poucos aspectos a serem observados são:

APIs de tempo de execução

Certas APIs de ML distribuídas só estão disponíveis dentro do Container Runtime, e a tentativa de importá-las fora do ambiente do Container Runtime falhará. Essas APIs estão disponíveis dentro de trabalhos de ML, mas precisam ser importadas dentro da carga útil do trabalho de ML.

# Attempting to import distributed runtime APIs in local/external
# environments will fail!
from snowflake.ml.modeling.distributors.xgboost import XGBEstimator

from snowflake.ml.jobs import remote

@remote(...)
def my_remote_function(...):
  # Move imports *inside* your ML Job payloads
  from snowflake.ml.modeling.distributors.xgboost import XGBEstimator  # This works!
  ...

job = my_remote_function()  # Start ML Job
job.wait()  # Wait for job to complete

Dimensionamento de clusters

A API scale_cluster() só funciona dentro de notebooks e não funcionará dentro de trabalhos de ML. Em vez disso, especifique o tamanho desejado do cluster no momento do envio do trabalho. Consulte Trabalhos de ML de vários nós do Snowflake para obter mais informações.

from snowflake.ml.jobs import remote

@remote(..., target_instances=4)
def my_remote_function(...):
  # 4-node cluster will be provisioned for distributed processing
  # inside this job. The cluster will be automatically cleaned up on
  # job termination.

Orquestração de pipelines¶

Depois de preparar seu pipeline de ponta a ponta, coloque-o em operação usando um orquestrador como gráficos de tarefas do Snowflake, notebooks programados ou orquestradores externos como o Airflow. O uso de uma estrutura de orquestração oferece várias vantagens importantes:

• Tolerância a falhas e confiabilidade por meio de novas tentativas automáticas e isolamento de falhas

• Observabilidade com histórico de execução, status em tempo real e alertas

• Programação e coordenação para gráficos de dependência complexos e vários acionadores

• Higienização operacional com integração de controle de versão e gerenciamento de configuração

O Snowflake ML é compatível com a maioria das estruturas de orquestração, incluindo Airflow, Dagster e Prefect. Se você já tem uma configuração de fluxo de trabalho/DAG existente, recomendamos simplesmente integrar seus fluxos de trabalho existentes aos recursos do Snowflake ML e transferir etapas intensivas em computação ou dados para trabalhos de ML ou UDFs. Se você não tiver uma configuração DAG, poderá usar gráficos de tarefas do Snowflake para uma solução nativa do Snowflake.

Para configurar a orquestração com um DAG no Snowflake, siga estas etapas de alto nível:

1. Prepare seu código de pipeline local de acordo com Preparar código

2. Crie um arquivo dag.py (ou qualquer outro nome) para manter sua definição de DAG

3. Implemente o formulário DAG do seu pipeline de acordo com este guia

4. Execute o script dag.py para implantar o gráfico de tarefa em sua conta Snowflake

Separação do desenvolvimento da produção¶

Recomendamos parametrizar seu script DAG para oferecer suporte ao isolamento de seus ambientes de desenvolvimento (DEV) e produção (PROD). Você pode usar o gerenciamento de conexões Snowflake, configurações específicas de aplicativos ou qualquer combinação dos dois para conseguir isso. O nível de isolamento necessário depende de seus requisitos de governança, mas geralmente recomendamos o uso de bancos de dados separados para DEV e PROD, onde o banco de dados do PROD é protegido por políticas RBAC que limitam o acesso a administradores e contas de serviço especializadas.

CI/CD¶

Você pode automatizar a validação e a implantação de seus pipelines usando pipelines de CI/CD, como Azure Pipelines e GitHub Actions. Em geral, recomendamos testar em um ambiente DEV ou STAGING antes de implantar no PROD. A prática recomendada é configurar seu repositório de controle de origem com portas de mesclagem que validam as alterações de código em DEV antes de fundir em sua ramificação de produção. As alterações na ramificação de produção podem ser implementadas no PROD continuamente (ou seja, para cada mudança) ou em alguma cadência regular (diária/semanal). A prática recomendada é executar uma validação final do estado da ramificação de produção em um ambiente DEV ou STAGING antes de implementar alterações no PROD. Use recursos da plataforma, como implantações e ambientes do GitHub Actions, para definir e configurar conexões com cada ambiente de implantação. Configure seu pipeline de CI/CD do para enviar suas alterações por push para o ambiente de implantação, incluindo:

• (Opcional) Criação de bibliotecas e módulos como pacotes Python e envio por push para um feed de pacote proprietário

• (Opcional) Carregamento de arquivos para uma área de preparação do Snowflake

  Isso é mais comumente necessário quando você utiliza snowflake.ml.jobs.submit_from_stage() em seu pipeline

  Como alternativa, você pode usar a integração com GitHub do Snowflake do para rastrear diretamente seu repositório do GitHub como uma área de preparação do Snowflake

• Execução de dag.py para implantar o gráfico de tarefas no ambiente configurado

• (Opcional) Acione e monitore a execução do gráfico de tarefas recém-implantado para verificar a validade