Snowflake Datasets¶

Installation¶

Das Dataset Python SDK ist in Snowpark ML (Python-Paket snowflake-ml-python) ab Version 1.5.0 enthalten. Eine Installationsanleitung finden Sie unter Lokale Verwendung von Snowflake ML.

Erforderliche Berechtigungen¶

Zum Erstellen von Datensets ist die Berechtigung CREATE DATASET auf Schemaebene erforderlich. Zum Bearbeiten von Datensätzen, z. B. zum Hinzufügen oder Löschen von Datenset-Versionen, ist die Berechtigung OWNERSHIP für das Datenset erforderlich. Zum Lesen eines Datensets benötigen Sie nur die Berechtigung USAGE für das Datenset (oder OWNERSHIP). Weitere Informationen über die Erteilung von Berechtigungen in Snowflake finden Sie unter GRANT <Berechtigungen>.

Erstellen und Verwenden von Datasets¶

Datasets werden erstellt, indem ein Snowpark-DataFrame an die Funktion snowflake.ml.dataset.create_from_dataframe übergeben wird.

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)

Datensets sind versioniert. Jede Version ist ein unveränderlicher, zeitpunktbezogener Snapshot der vom Dataset verwalteten Daten. Die Python-API enthält die Eigenschaft Dataset.selected_version, die angibt, ob ein bestimmtes Datenset zur Verwendung ausgewählt ist. Diese Eigenschaft wird automatisch von den Factory-Methoden dataset.create_from_dataframe und dataset.load_dataset gesetzt, sodass beim Erstellen eines Datensets automatisch die erstellte Version ausgewählt wird. Die Methoden Dataset.select_version und Dataset.create_version können auch verwendet werden, um explizit zwischen Versionen zu wechseln. Das Lesen aus einem Datenset erfolgt aus der aktiven ausgewählten Version.

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

Lesen von Daten aus Datenset¶

Die Daten der Dataset-Version werden als gleichmäßig große Dateien im Apache Parquet-Format gespeichert. Die Klasse Dataset bietet eine API ähnlich der von FileSet zum Lesen von Daten aus Snowflake Datasets, einschließlich integrierter Konnektoren für TensorFlow und PyTorch. Die API ist erweiterbar, um kundenspezifische Framework-Konnektoren zu unterstützen.

Das Lesen aus einem Datenset erfordert eine aktive ausgewählte Version.

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

Aktuelle Beschränkungen und bekannte Probleme¶

• Dataset-Namen sind SQL-Bezeichner und unterliegen den Snowflake-Anforderungen an Bezeichner.

• Dataset-Versionen sind String-Werte mit einer maximale Länge von 128 Zeichen. Einige Zeichen sind nicht zulässig und führen zu einer Fehlermeldung.

• Bestimmte Abfrageoperationen auf Datasets mit umfangreichen Schemas (mehr als etwa 4.000 Spalten) sind nicht vollständig optimiert. Dies wird sich in kommenden Releases verbessern.