Partilhar via

Carregar dados usando o Mosaic Streaming

  • Artigo

Este artigo descreve como usar o Mosaic Streaming para converter dados do Apache Spark para um formato compatível com o PyTorch.

Mosaic Streaming é uma biblioteca de carregamento de dados de código aberto. Ele permite treinamento de nó único ou distribuído e avaliação de modelos de aprendizado profundo de conjuntos de dados que já estão carregados como Apache Spark DataFrames. O Mosaic Streaming suporta principalmente o Mosaic Composer, mas também se integra com o nativo PyTorch, o PyTorch Lightning e o TorchDistributor. O Mosaic Streaming oferece uma série de benefícios em relação aos DataLoaders PyTorch tradicionais, incluindo:

  • Compatibilidade com qualquer tipo de dados, incluindo imagens, texto, vídeo e dados multimodais.
  • Suporte para os principais provedores de armazenamento em nuvem (AWS, OCI, GCS, Azure, Databricks UC Volume e qualquer armazenamento de objetos compatível com o S3, como Cloudflare R2, Coreweave, Backblaze b2, etc.)
  • Maximizando as garantias de correção, desempenho, flexibilidade e facilidade de uso. Para obter mais informações, consulte a página de recursos principais.

Para obter informações gerais sobre o Mosaic Streaming, consulte a documentação da API de streaming.

Nota

O Mosaic Streaming foi pré-instalado em todas as versões do Databricks Runtime 15.2 ML e superior.

Carregar dados do Spark DataFrames usando o Mosaic Streaming

O Mosaic Streaming fornece um fluxo de trabalho simples para a conversão do Apache Spark para o formato Mosaic Data Shard (MDS), que pode ser carregado para uso em um ambiente distribuído.

O fluxo de trabalho recomendado é:

  1. Use o Apache Spark para carregar e, opcionalmente, pré-processar dados.
  2. Use streaming.base.converters.dataframe_to_mds para salvar o dataframe em disco para armazenamento transitório e/ou em um volume do Catálogo Unity para armazenamento persistente. Esses dados serão armazenados no formato MDS e podem ser otimizados com suporte para compressão e hashing. Casos de uso avançados também podem incluir o pré-processamento de dados usando UDFs. Veja o tutorial do Spark DataFrame to MDS para obter mais informações.
  3. Use streaming.StreamingDataset para carregar os dados necessários na memória. StreamingDataset é uma versão do IterableDataset do PyTorch que apresenta embaralhamento elasticamente determinístico, que permite uma retomada rápida no meio da época. Consulte a documentação do StreamingDataset para obter mais informações.
  4. Use streaming.StreamingDataLoader para carregar os dados necessários para treinamento/avaliação/teste. StreamingDataLoader é uma versão do DataLoader do PyTorch que fornece uma interface adicional de ponto de verificação/retomada, para a qual rastreia o número de amostras vistas pelo modelo nesta classificação.

Para obter um exemplo completo, consulte o seguinte bloco de anotações:

Simplifique o carregamento de dados do Spark para o PyTorch usando o notebook Mosaic Streaming

Obter bloco de notas

Resolução de problemas: Erro de autenticação

Se você vir o seguinte erro ao carregar dados de um volume do Catálogo Unity usando StreamingDataset, configure as variáveis de ambiente conforme mostrado abaixo.

ValueError: default auth: cannot configure default credentials, please check https://docs.databricks.com/en/dev-tools/auth.html#databricks-client-unified-authentication to configure credentials for your preferred authentication method.

Nota

Se você vir esse erro ao executar o treinamento distribuído usando TorchDistributor, você também deve definir as variáveis de ambiente nos nós de trabalho.

db_host = "https://your-databricks-host.databricks.com"
db_token = "YOUR API TOKEN" # Create a token with either method from https://docs.databricks.com/en/dev-tools/auth/index.html#databricks-authentication-methods

def your_training_function():
  import os
  os.environ['DATABRICKS_HOST'] = db_host
  os.environ['DATABRICKS_TOKEN'] = db_token

# The above function can be distributed with TorchDistributor:
# from pyspark.ml.torch.distributor import TorchDistributor
# distributor = TorchDistributor(...)
# distributor.run(your_training_function)