Petastorm ile veri yükleme (kullanım dışı)

Petastorm, derin öğrenme modellerinin tek düğümlü veya dağıtılmış eğitimine olanak tanıyan açık kaynak bir veri erişim kitaplığıdır. Bu kitaplık doğrudan Apache Parquet biçimindeki veri kümelerinden ve Apache Spark DataFrame olarak yüklenen veri kümelerinden eğitim sağlar. Petastorm, TensorFlow ve PyTorch gibi popüler eğitim çerçevelerini destekler.

Petastorm hakkında daha fazla bilgi için Petastorm GitHub sayfasını veya Petastorm API belgelerini ziyaret edebilirsiniz.

Önkoşullar

Not

Azure Synapse GPU özellikli havuzlar için Önizleme artık kullanım dışı bırakıldı.

Apache Spark oturumunu yapılandırma

Oturumun başında birkaç Apache Spark ayarı yapılandırmamız gerekir. Çoğu durumda, yalnızca numExecutors ve spark.rapids.memory.gpu.reserve'yi ayarlamamız gerekir. Örnekte, %%configure komutu ile Spark yapılandırmalarının nasıl geçirilebileceğini görebilirsiniz. Her parametrenin ayrıntılı anlamı Apache Spark yapılandırma belgelerinde açıklanmıştır.

%%configure -f
{
    "numExecutors": 3,
    "conf":{
        "spark.rapids.memory.gpu.reserve": "10g"
   }
}

Petastorm yazım API'leri

Petastorm kullanılarak oluşturulan bir veri kümesi Apache Parquet biçiminde depolanır. Parquet şemasının üzerinde Petastorm, çok boyutlu dizileri Petastorm veri kümesinin yerel bir parçası haline getiren daha üst düzey şema bilgilerini de depolar.

Örnekte PySpark kullanarak bir veri kümesi oluşturacağız. Veri kümesini bir Azure Data Lake Storage 2. Nesil hesabına yazarız.

import numpy as np
from pyspark.sql import SparkSession
from pyspark.sql.types import IntegerType

from petastorm.codecs import ScalarCodec, CompressedImageCodec, NdarrayCodec
from petastorm.etl.dataset_metadata import materialize_dataset
from petastorm.unischema import dict_to_spark_row, Unischema, UnischemaField

# The schema defines how the dataset schema looks like
HelloWorldSchema = Unischema('HelloWorldSchema', [
    UnischemaField('id', np.int32, (), ScalarCodec(IntegerType()), False),
    UnischemaField('image1', np.uint8, (128, 256, 3), CompressedImageCodec('png'), False),
    UnischemaField('array_4d', np.uint8, (None, 128, 30, None), NdarrayCodec(), False),
])


def row_generator(x):
    """Returns a single entry in the generated dataset. Return a bunch of random values as an example."""
    return {'id': x,
            'image1': np.random.randint(0, 255, dtype=np.uint8, size=(128, 256, 3)),
            'array_4d': np.random.randint(0, 255, dtype=np.uint8, size=(4, 128, 30, 3))}


def generate_petastorm_dataset(output_url):
    rowgroup_size_mb = 256

    spark = SparkSession.builder.config('spark.driver.memory', '2g').master('local[2]').getOrCreate()
    sc = spark.sparkContext

    # Wrap dataset materialization portion. Will take care of setting up spark environment variables as
    # well as save petastorm specific metadata
    rows_count = 10
    with materialize_dataset(spark, output_url, HelloWorldSchema, rowgroup_size_mb):

        rows_rdd = sc.parallelize(range(rows_count))\
            .map(row_generator)\
            .map(lambda x: dict_to_spark_row(HelloWorldSchema, x))

        spark.createDataFrame(rows_rdd, HelloWorldSchema.as_spark_schema()) \
            .coalesce(10) \
            .write \
            .mode('overwrite') \
            .parquet(output_url)


output_url = 'abfs://container_name@storage_account_url/data_dir' #use your own adls account info
generate_petastorm_dataset(output_url)

Petastorm okuma API'leri

Birincil depolama hesabından veri kümesini okuma

petastorm.reader.Reader sınıfı, TensorFlow veya PyTorch gibi bir ML çerçevesinden verilere erişen kullanıcı kodu için ana giriş noktasıdır. petastorm.reader.Reader sınıfı ve petastorm.make_reader fabrika yöntemi kullanılarak bir veri kümesi okuyabilirsiniz.

Örnekte, URL abfs protokollerini nasıl geçirebileceğinizi görebilirsiniz.

from petastorm import make_reader

#on primary storage associated with the workspace, url can be abbreviated with container path for data directory
with make_reader('abfs://<container_name>/<data directory path>/') as reader:
    for row in reader:
        print(row)

İkincil depolama hesabından veri kümesini okuma

Alternatif bir depolama hesabı kullanıyorsanız, bağlı hizmeti otomatik olarak kimlik doğrulaması yapmak ve hesaptan okumak için ayarladığınızdan emin olun. Ayrıca, şu özellikleri değiştirmeniz gerekir: remote_url, account_nameve linked_service_name.

from petastorm import make_reader

# create sas token for storage account access, use your own adls account info
remote_url = "abfs://container_name@storage_account_url"
account_name = "<<adls account name>>"
linked_service_name = '<<linked service name>>'
TokenLibrary = spark._jvm.com.microsoft.azure.synapse.tokenlibrary.TokenLibrary
sas_token = TokenLibrary.getConnectionString(linked_service_name)

with make_reader('{}/data_directory'.format(remote_url), storage_options = {'sas_token' : sas_token}) as reader:
    for row in reader:
        print(row)

Veri kümesini toplu olarak okuma

Örnekte, verileri toplu olarak okumak için url abfs protokollerini nasıl geçirebileceğinizi görebilirsiniz. Bu örnekte sınıfı kullanılır make_batch_reader .

from petastorm import make_batch_reader

with make_batch_reader('abfs://<container_name>/<data directory path>/', schema_fields=["value1", "value2"]) as reader:
    for schema_view in reader:
        print("Batched read:\nvalue1: {0} value2: {1}".format(schema_view.value1, schema_view.value2))

PyTorch API'si

PyTorch'tan bir Petastorm veri kümesini okumak için bağdaştırıcı petastorm.pytorch.DataLoader sınıfını kullanabilirsiniz. Bu bağdaştırıcı, özel PyTorch harmanlama işlevlerinin ve dönüşümlerinin sağlanmasını sağlar.

Bu örnekte, Petastorm DataLoader'ın make_reader API yardımıyla petastorm veri kümesini yüklemek için nasıl kullanılabileceğini göstereceğiz. Bu ilk bölüm, bir Net sınıfın ve traintest işlevinin tanımını oluşturur.

from __future__ import division, print_function

import argparse
import pyarrow
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torchvision import transforms

from petastorm import make_reader, TransformSpec
from petastorm.pytorch import DataLoader
from pyspark.sql.functions import col

class Net(nn.Module):
    def __init__(self):
        super(Net, self).__init__()
        self.conv1 = nn.Conv2d(1, 10, kernel_size=5)
        self.conv2 = nn.Conv2d(10, 20, kernel_size=5)
        self.conv2_drop = nn.Dropout2d()
        self.fc1 = nn.Linear(320, 50)
        self.fc2 = nn.Linear(50, 10)

    def forward(self, x):
        x = F.relu(F.max_pool2d(self.conv1(x), 2))
        x = F.relu(F.max_pool2d(self.conv2_drop(self.conv2(x)), 2))
        x = x.view(-1, 320)
        x = F.relu(self.fc1(x))
        x = F.dropout(x, training=self.training)
        x = self.fc2(x)
        return F.log_softmax(x, dim=1)

def train(model, device, train_loader, log_interval, optimizer, epoch):
    model.train()
    for batch_idx, row in enumerate(train_loader):
        data, target = row['image'].to(device), row['digit'].to(device)
        optimizer.zero_grad()
        output = model(data)
        loss = F.nll_loss(output, target)
        loss.backward()
        optimizer.step()
        if batch_idx % log_interval == 0:
            print('Train Epoch: {} [{}]\tLoss: {:.6f}'.format(
                epoch, batch_idx * len(data), loss.item()))

def test(model, device, test_loader):
    model.eval()
    test_loss = 0
    correct = 0
    count = 0
    with torch.no_grad():
        for row in test_loader:
            data, target = row['image'].to(device), row['digit'].to(device)
            output = model(data)
            test_loss += F.nll_loss(output, target, reduction='sum').item()  # sum up batch loss
            pred = output.max(1, keepdim=True)[1]  # get the index of the max log-probability
            correct += pred.eq(target.view_as(pred)).sum().item()
            count += data.shape[0]
    test_loss /= count
    print('\nTest set: Average loss: {:.4f}, Accuracy: {}/{} ({:.0f}%)\n'.format(
        test_loss, correct, count, 100. * correct / count))

def _transform_row(mnist_row):
    # For this example, the images are stored as simpler ndarray (28,28), but the
    # training network expects 3-dim images, hence the additional lambda transform.
    transform = transforms.Compose([
        transforms.Lambda(lambda nd: nd.reshape(28, 28, 1)),
        transforms.ToTensor(),
        transforms.Normalize((0.1307,), (0.3081,))
    ])
    # In addition, the petastorm pytorch DataLoader does not distinguish the notion of
    # data or target transform, but that actually gives the user more flexibility
    # to make the desired partial transform, as shown here.
    result_row = {
        'image': transform(mnist_row['image']),
        'digit': mnist_row['digit']
    }

    return result_row

Bu örnekte ara verileri depolamak için bir Azure Data Lake Storage hesabı kullanılır. Bu verileri depolamak için depolama hesabına bir Bağlı Hizmet ayarlamanız ve şu bilgi parçalarını almanız gerekir: remote_url, account_nameve linked_service_name.

from petastorm import make_reader

# create sas token for storage account access, use your own adls account info
remote_url = "abfs://container_name@storage_account_url"
account_name = "<account name>"
linked_service_name = '<linked service name>'
TokenLibrary = spark._jvm.com.microsoft.azure.synapse.tokenlibrary.TokenLibrary
sas_token = TokenLibrary.getConnectionString(linked_service_name)

# Read Petastorm dataset and apply custom PyTorch transformation functions 

device = torch.device('cpu') #For GPU, it will be torch.device('cuda'). More details: https://pytorch.org/docs/stable/tensor_attributes.html#torch-device

model = Net().to(device)
optimizer = optim.SGD(model.parameters(), lr=0.01, momentum=0.5)

loop_epochs = 1
reader_epochs = 1

transform = TransformSpec(_transform_row, removed_fields=['idx'])

for epoch in range(1, loop_epochs + 1):
    with DataLoader(make_reader('{}/train'.format(remote_url), num_epochs=reader_epochs, transform_spec=transform),batch_size=5) as train_loader:
        train(model, device, train_loader, 10, optimizer, epoch)
    with DataLoader(make_reader('{}/test'.format(remote_url), num_epochs=reader_epochs, transform_spec=transform), batch_size=5) as test_loader:
        test(model, device, test_loader)

Sonraki adımlar

  • Last updated on