データのロードと書き込み¶

from snowflake.ml.data.data_connector import DataConnector from snowflake.snowpark.context import get_active_session session = get_active_session() # Create DataConnector from a Snowflake table data_connector = DataConnector.from_dataframe(session.table("example-table-name")) 

from snowflake.ml.data.data_connector import DataConnector # Create DataConnector from a Snowflake Dataset data_connector = DataConnector.from_dataset(snowflake_dataset) 

from snowflake.ml.data.data_connector import DataConnector from snowflake.ml.modeling.distributors.xgboost.xgboost_estimator import ( XGBEstimator, XGBScalingConfig, ) from snowflake.snowpark.context import get_active_session session = get_active_session() # Create DataConnector from a Snowpark dataframe snowflake_df = session.table("TRAINING_TABLE") data_connector = DataConnector.from_dataframe(snowflake_df) # Create Snowflake XGBoost estimator snowflake_est = XGBEstimator( n_estimators=1, objective="reg:squarederror", scaling_config=XGBScalingConfig(use_gpu=False), ) # Train using the data connector # When using a data connector, input_cols and label_col must be provided fit_booster = snowflake_est.fit( data_connector, input_cols=NUMERICAL_COLS, label_col=LABEL_COL ) 

from sklearn import datasets from snowflake.ml.data.sharded_data_connector import ShardedDataConnector from snowflake.ml.modeling.pytorch import ( PyTorchTrainer, ScalingConfig, WorkerResourceConfig, getContext, ) from torch import nn from snowflake.snowpark.context import get_active_session session = get_active_session() # Create the Snowflake data from a Snowpark dataframe digits = datasets.load_digits(as_frame=True).frame digits_df = session.create_dataframe(digits) # Create sharded data connector sharded_data_connector = ShardedDataConnector.from_dataframe(digits_df) # Define the PyTorch model class DigitsModel(nn.Module): def __init__(self): super(DigitsModel, self).__init__() self.flatten = nn.Flatten() self.linear_relu_stack = nn.Sequential( nn.Linear(8 * 8, 512), nn.ReLU(), nn.Linear(512, 512), nn.ReLU(), nn.Linear(512, 10) ) def forward(self, x): x = self.flatten(x) logits = self.linear_relu_stack(x) return logits # Define training function that runs across multiple nodes or devices # Each process receives a unique data shard def train_func(): import os import torch import torch.distributed as dist from torch.utils.data import DataLoader from torch import nn from torch.nn.parallel import DistributedDataParallel as DDP # Get context with data shards and model directory context = getContext() dataset_map = context.get_dataset_map() model_dir = context.get_model_dir() training_data = dataset_map["train"].get_shard().to_torch_dataset() train_dataloader = DataLoader(training_data, batch_size=batch_size, drop_last=True) dist.init_process_group() device = "cpu" label_col = '"target"' batch_size = 64 model = DDP(DigitsModel()) loss_fn = nn.CrossEntropyLoss() optimizer = torch.optim.SGD(model.parameters(), lr=1e-3) # Training loop for epoch in range(5): for batch, batch_data in enumerate(train_dataloader): y = batch_data.pop(label_col).flatten().type(torch.LongTensor).to(device) X = torch.concat( [tensor.to(torch.float32) for tensor in batch_data.values()], dim=-1, ).to(device) pred = model(X) loss = loss_fn(pred, y) optimizer.zero_grad() loss.backward() optimizer.step() if batch % 100 == 0: print(f"Epoch {epoch}, Batch {batch}, Loss: {loss.item()}") # Save the model if dist.get_rank() == 0: torch.save(model.state_dict(), os.path.join(model_dir, "digits_model.pth")) # Create PyTorch trainer with scaling configuration pytorch_trainer = PyTorchTrainer( train_func=train_func, scaling_config=ScalingConfig( num_nodes=1, num_workers_per_node=4, resource_requirements_per_worker=WorkerResourceConfig(num_cpus=1, num_gpus=0), ), ) # Run distributed training response = pytorch_trainer.run( dataset_map=dict( train=sharded_data_connector, ) ) 

import ray from snowflake.ml.ray.datasource.stage_parquet_file_datasource import SFStageParquetDataSource from snowflake.ml.data.data_connector import DataConnector data_source = SFStageParquetDataSource( stage_location="@stage/path/", database="DB_NAME", # optional schema="SCHEMA_NAME", # optional file_pattern='*.parquet', # optional ) # Build Ray dataset from provided datasources ray_ds = ray.data.read_datasource(data_source) dc = DataConnector.from_ray_dataset(ray_ds)