Ensembles de données Snowflake¶

Installation¶

Le SDK Dataset Python est inclus dans Snowpark ML (paquet Python snowflake-ml-python) à partir de la version 1.5.0. Pour des instructions d’installation, voir Utilisation de Snowflake ML en local.

Privilèges requis¶

La création d’ensembles de données requiert le privilège CREATE DATASET au niveau du schéma. La modification des ensembles de données, par exemple l’ajout ou la suppression de versions d’ensembles de données, nécessite OWNERSHIP sur l’ensemble de données. La lecture d’un ensemble de données ne nécessite que le privilège USAGE sur l’ensemble de données (ou OWNERSHIP). Pour plus d’informations sur l’octroi de privilèges dans Snowflake voir GRANT <privilèges>.

Création et utilisation d’ensembles de données¶

Les ensembles de données sont créés en transmettant une adresse Snowpark DataFrame à la fonction 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)

Les ensembles de données ont plusieurs versions. Chaque version est un instantané immuable et ponctuel des données gérées par l’ensemble de données. L’API Python comprend une propriété Dataset.selected_version qui indique si un ensemble de données donné est sélectionné pour être utilisé. Cette propriété est automatiquement définie par les méthodes d’usine dataset.create_from_dataframe et dataset.load_dataset, de sorte que la création d’un ensemble de données sélectionne automatiquement la version créée. Les méthodes Dataset.select_version et Dataset.create_version peuvent également être utilisées pour passer explicitement d’une version à l’autre. La lecture à partir d’un ensemble de données lit à partir de la version active sélectionnée.

# 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"

Lecture de données à partir d’ensembles de données¶

Les données de la version de l’ensemble de données sont stockées sous forme de fichiers de taille égale au format Apache Parquet. La classe Dataset fournit une API similaire à celle de FileSet pour la lecture de données à partir d’ensembles de données Snowflake, y compris des connecteurs intégrés pour TensorFlow et PyTorch. L’API est extensible pour prendre en charge des connecteurs de cadre personnalisés.

La lecture d’un ensemble de données nécessite une version sélectionnée active.

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())

Limites et problèmes connus actuels¶

• Les noms des ensembles de données sont des identificateurs SQL et sont soumis aux exigences de l’identificateur Snowflake.

• Les versions des ensembles de données sont des chaînes et ont une longueur maximale de 128 caractères. Certains caractères ne sont pas autorisés et produiront un message d’erreur.

• Certaines opérations de requête sur des ensembles de données ayant des schémas étendus (plus de 4 000 colonnes environ) ne sont pas entièrement optimisées. Cela devrait s’améliorer dans les prochaines versions.