Conjuntos de dados do Snowflake¶

Instalação¶

O conjunto de dados Python SDK está incluído no Snowpark ML (pacote Python snowflake-ml-python) a partir da versão 1.5.0. Para instruções de instalação, consulte Como usar o Snowflake ML localmente.

Privilégios obrigatórios¶

A criação de conjuntos de dados requer o privilégio CREATE DATASET no nível de esquema. A modificação de conjuntos de dados, por exemplo, adicionando ou excluindo versões de conjuntos de dados, exige OWNERSHIP no conjunto de dados. A leitura de um conjunto de dados requer apenas o privilégio USAGE no conjunto de dados (ou OWNERSHIP). Para obter mais informações sobre como conceder privilégios no Snowflake, consulte GRANT <privilégios>.

Criação e uso do conjuntos de dados¶

Os conjuntos de dados são criados passando um Snowpark DataFrame para a função snowflake.ml.dataset.create_from_dataframe.

from snowflake import snowpark
from snowflake.ml import dataset

# Create Snowpark Session
# See https://docs.snowflake.com/en/developer-guide/snowpark/python/creating-session
session = snowpark.Session.builder.configs(connection_parameters).create()

# Create a Snowpark DataFrame to serve as a data source
# In this example, we generate a random table with 100 rows and 1 column
df = session.sql(
    "select uniform(0, 10, random(1)) as x, uniform(0, 10, random(2)) as y from table(generator(rowcount => 100))"
)

# Materialize DataFrame contents into a Dataset
ds1 = dataset.create_from_dataframe(
    session,
    "my_dataset",
    "version1",
    input_dataframe=df)

Os conjuntos de dados possuem versão. Cada versão é um instantâneo imutável e pontual dos dados gerenciados pelo conjunto de dados. A API de Python inclui uma propriedade Dataset.selected_version que indica se um determinado conjunto de dados está selecionado para uso. Essa propriedade é definida automaticamente pelos métodos de fábrica dataset.create_from_dataframe e dataset.load_dataset, portanto, a criação de um conjunto de dados seleciona automaticamente a versão criada. Os métodos Dataset.select_version e Dataset.create_version também podem ser usados para alternar explicitamente entre versões. A leitura a partir de um conjunto de dados lê a versão selecionada ativa.

# Inspect currently selected version
print(ds1.selected_version) # DatasetVersion(dataset='my_dataset', version='version1')
print(ds1.selected_version.created_on) # Prints creation timestamp

# List all versions in the Dataset
print(ds1.list_versions()) # ["version1"]

# Create a new version
ds2 = ds1.create_version("version2", df)
print(ds1.selected_version.name)  # "version1"
print(ds2.selected_version.name)  # "version2"
print(ds1.list_versions())        # ["version1", "version2"]

# selected_version is immutable, meaning switching versions with
# ds1.select_version() returns a new Dataset object without
# affecting ds1.selected_version
ds3 = ds1.select_version("version2")
print(ds1.selected_version.name)  # "version1"
print(ds3.selected_version.name)  # "version2"

Leitura de dados de conjuntos de dados¶

Os dados da versão do conjunto de dados são armazenados como arquivos de tamanho uniforme no formato Apache Parquet. A classe Dataset fornece uma API semelhante ao de FileSet para leitura de dados de conjuntos de dados Snowflake, incluindo conectores integrados para TensorFlow e PyTorch. A API é extensível para oferecer suporte a conectores de estrutura personalizados.

Ler um conjunto de dados requer uma versão selecionada ativa.

import tensorflow as tf

# Convert Snowflake Dataset to TensorFlow Dataset
tf_dataset = ds1.read.to_tf_dataset(batch_size=32)

# Train a TensorFlow model
for batch in tf_dataset:
    # Extract and build tensors as needed
    input_tensor = tf.stack(list(batch.values()), axis=-1)

    # Forward pass (details not included for brevity)
    outputs = model(input_tensor)

import torch

# Convert Snowflake Dataset to PyTorch DataPipe
pt_datapipe = ds1.read.to_torch_datapipe(batch_size=32)

# Train a PyTorch model
for batch in pt_datapipe:
    # Extract and build tensors as needed
    input_tensor = torch.stack([torch.from_numpy(v) for v in batch.values()], dim=-1)

    # Forward pass (details not included for brevity)
    outputs = model(input_tensor)

from snowflake.ml.modeling.ensemble import random_forest_regressor

# Get a Snowpark DataFrame
ds_df = ds1.read.to_snowpark_dataframe()

# Note ds_df != df
ds_df.explain()
df.explain()

# Train a model in Snowpark ML
xgboost_model = random_forest_regressor.RandomForestRegressor(
    n_estimators=100,
    random_state=42,
    input_cols=["X"],
    label_cols=["Y"],
)
xgboost_model.fit(ds_df)

print(ds1.read.files()) # ['snow://dataset/my_dataset/versions/version1/data_0_0_0.snappy.parquet']

import pyarrow.parquet as pq
pd_ds = pq.ParquetDataset(ds1.read.files(), filesystem=ds1.read.filesystem())

import dask.dataframe as dd
dd_df = dd.read_parquet(ds1.read.files(), filesystem=ds1.read.filesystem())

Limitações atuais e problemas conhecidos¶

• Os nomes dos conjuntos de dados são identificadores SQL e sujeitos a requisitos do identificador Snowflake.

• As versões do conjunto de dados são cadeias de caracteres e têm um comprimento máximo de 128 caracteres. Alguns caracteres não são permitidos e produzirão uma mensagem de erro.

• Certas operações de consulta em conjuntos de dados com esquemas amplos (mais de 4.000 colunas) não são totalmente otimizadas. Isso deve melhorar nas próximas versões.