TorchData 0.3.0 Beta : チュートリアル (翻訳/解説)

翻訳 : (株)クラスキャット セールスインフォメーション
作成日時 : 03/15/2022

* 本ページは、TorchData の以下のドキュメントを翻訳した上で適宜、補足説明したものです：

• TorchData : Tutorial

* サンプルコードの動作確認はしておりますが、必要な場合には適宜、追加改変しています。
* ご自由にリンクを張って頂いてかまいませんが、sales-info@classcat.com までご一報いただけると嬉しいです。

 

 

TorchData 0.3.0 Beta : チュートリアル

DataPipe の使用

以下のステップで CSV ファイルからデータをロードしたいと仮定します :

• ディレクトリの総ての CSV ファイルをリストアップする。
• CSV ファイルをロードする。
• CSV ファイルをパースして行を yield する。

上記の操作に役立つ幾つかの 組み込み DataPipe があります。

• FileLister – ディレクトリのファイルをリストアップする。
• Filter – 与えられた関数に基づいて DataPipe の要素をフィルタリングする。
• FileOpener – ファイルパスを使ってオープンされたファイルストリームを返す。
• CSVParser – ファイルストリームを使用し、CSV コンテンツをパースし、そして一度にパースされた行を返す。

例として、CSVParser のソースコードはこのようなものです

@functional_datapipe("parse_csv")
class CSVParserIterDataPipe(IterDataPipe):
    def __init__(self, dp, **fmtparams) -> None:
        self.dp = dp
        self.fmtparams = fmtparams

    def __iter__(self) -> Iterator[Union[Str_Or_Bytes, Tuple[str, Str_Or_Bytes]]]:
        for path, file in self.source_datapipe:
            stream = self._helper.skip_lines(file)
            stream = self._helper.strip_newline(stream)
            stream = self._helper.decode(stream)
            yield from self._helper.return_path(stream, path=path)  # Returns 1 line at a time as List[str or bytes]

別のセクションで記述されたように、DataPipe はそれらの関数型 (推奨) かクラス・コンストラクタを使用して呼び出せます。パイプラインは以下のように組み立てることができます :

import torchdata.datapipes as dp

FOLDER = 'path/2/csv/folder'
datapipe = dp.iter.FileLister([FOLDER]).filter(filter_fn=lambda filename: filename.endswith('.csv'))
datapipe = dp.iter.FileOpener(datapipe, mode='rt')
datapipe = datapipe.parse_csv(delimiter=',')

for d in datapipe: # Iterating through the data
     pass

ここで組み込み IterDataPipes の完全なリストをそして ここで MapDataPipes のリストを見つけられます。

 

DataLoader との連携

このセクションでは DataPipe を DataLoader と共にどのように利用できるか実演します。殆どの場合、DataLoader への入力引数として dataset=datapipe を単に渡すことによりそれを利用することができるはずです。DataLoader に関する詳細なドキュメントは、このページ を見てください。

このサンプルについては、最初に、ランダムラベルとデータを持つ幾つかの CSV ファイルを生成するヘルパー関数を持ちます。

import csv
import random

def generate_csv(file_label, num_rows: int = 5000, num_features: int = 20) -> None:
    fieldnames = ['label'] + [f'c{i}' for i in range(num_features)]
    writer = csv.DictWriter(open(f"sample_data{file_label}.csv", "w"), fieldnames=fieldnames)
    writer.writerow({col: col for col in fieldnames})  # writing the header row
    for i in range(num_rows):
        row_data = {col: random.random() for col in fieldnames}
        row_data['label'] = random.randint(0, 9)
        writer.writerow(row_data)

次に、生成された CSV ファイルを読んでパースするために DataPipe を構築します :

import numpy as np
import torchdata.datapipes as dp

def build_datapipes(root_dir="."):
    datapipe = dp.iter.FileLister(root_dir)
    datapipe = datapipe.filter(filter_fn=lambda filename: "sample_data" in filename and filename.endswith(".csv"))
    datapipe = dp.iter.FileOpener(datapipe, mode='rt')
    datapipe = datapipe.parse_csv(delimiter=",", skip_lines=1)
    datapipe = datapipe.map(lambda row: {"label": np.array(row[0], np.int32),
                                         "data": np.array(row[1:], dtype=np.float64)})
    return datapipe

最後に、’__main__’ に総てをまとめて DataPipe を DataLoader に渡します。

from torch.utils.data import DataLoader

if __name__ == '__main__':
    num_files_to_generate = 3
    for i in range(num_files_to_generate):
        generate_csv(file_label=i)
    datapipe = build_datapipes()
    dl = DataLoader(dataset=datapipe, batch_size=50, shuffle=True)
    first = next(iter(dl))
    labels, features = first['label'], first['data']
    print(f"Labels batch shape: {labels.size()}")
    print(f"Feature batch shape: {features.size()}")

ラベルと特徴の単一バッチの shape を示すために以下のステートメントがプリントされます。

Labels batch shape: 50
Feature batch shape: torch.Size([50, 20])

このページ で様々な研究分野のためのより多くの DataPipe 実装サンプルを見つけることができます。

 

カスタム DataPipe の実装

現在、非常に多くの組み込み DataPipe を既に持ち、そしてそれらが殆どの必要なデータ処理操作をカバーすることを期待しています。もしそれらが貴方のニーズをサポートしないならば、独自のカスタム DataPipe を作成できます。

ガイド的なサンプルとして、入力イテレータに callable を適用する IterDataPipe を実装しましょう。MapDataPipe については、例として map フォルダを見てください、そして __iter__ メソッドの代わりに __getitem__ メソッドのために以下のステップに従ってください。

 

命名

DataPipe のための命名規則は “演算”-er で、IterDataPipe or MapDataPipe が続きます、これは各々の DataPipe が本質的にはソース DataPipe から生成されるデータに演算を適用するコンテナであるためです。簡潔にするため、init ファイルでは単に “Operation-er” とエイリアスします。IterDataPipe サンプルについては、モジュール MapperIterDataPipe と命名してそれを torchdata.datapipes 下の iter.Mapper としてエイリアスを付けます。

 

コンストラクタ

DataSet は今では DataPipe のスタックとして一般に構築されますので、各 DataPipe は典型的にはその最初の引数としてソース DataPipe を取ります。ここには例として Mapper の単純化バージョンがあります :

from torchdata.datapipes.iter import IterDataPipe

class MapperIterDataPipe(IterDataPipe):
    def __init__(self, source_dp: IterDataPipe, fn) -> None:
        super().__init__()
        self.source_dp = source_dp
        self.fn = fn

Note :

• 遅延データロードをサポートしてメモリを節約するために、__init__ 関数でソース DataPipe からデータをロードすることは回避します。
  

• IterDataPipe インスタンスがメモリにデータを保持する場合、データの in-place 変更に注意してください。インスタンスから 2 番目のイテレータが作成されたとき、データは既に変更されているかもしれません。各イテレータについてデータを deepcopy するために IterableWrapper クラスを参照してください。
  

• 既存の DataPipe の関数名から取られる変数名は回避してください。例えば、.filter は FilterIterDataPipe を呼び出すために使用できる関数名です。別の IterDataPipe 内に filter という名前の変数を持つことは混乱に繋がる可能性があります。

 

イテレータ

IterDataPipes について、ソース IterDataPipe からデータを消費するためには __iter__ 関数が必要です、そして yeild する前にデータに対して演算を適用します。

class MapperIterDataPipe(IterDataPipe):
    # ... See __init__() defined above

    def __iter__(self):
        for d in self.dp:
            yield self.fn(d)

 

長さ

多くの場合、MapperIterDataPipe サンプルでのように、DataPipe の __len__ メソッドはソース DataPipe の長さを返します。

class MapperIterDataPipe(IterDataPipe):
    # ... See __iter__() defined above

    def __len__(self):
        return len(self.dp)

けれども、__len__ は IterDataPipe のオプションであり勧められない場合も多いことに注意してください。下の DataPipe の使用セクションの CSVParserIterDataPipe については、__len__ は実装されていません、何故ならば各ファイルの行数はそれをロードする前には未知であるためです。幾つかの特殊なケースでは、__len__ は入力に依存して整数を返すか Error を上げるのいずれかをさせることができます。それらの場合には、list(dp) のような Python の組み込み関数をサポートするために Error は TypeError である必要があります。

 

関数型 API で DataPipe を登録する

各 DataPipe はデコレータ functional_datapipe を使用して関数型呼び出しをサポートするために登録することができます。

@functional_datapipe("map")
class MapperIterDataPipe(IterDataPipe):
   # ...

そして DataPipe のスタックはそれらの関数型形式 (推奨) かクラス・コンストラクタを使用して構築できます :

import torchdata.datapipes as dp

# Using functional form (recommended)
datapipes1 = dp.iter.FileOpener(['a.file', 'b.file']).map(fn=decoder).shuffle().batch(2)
# Using class constructors
datapipes2 = dp.iter.FileOpener(['a.file', 'b.file'])
datapipes2 = dp.iter.Mapper(datapipes2, fn=decoder)
datapipes2 = dp.iter.Shuffler(datapipes2)
datapipes2 = dp.iter.Batcher(datapipes2, 2)

上のサンプルで、datapipes1 と datapipes2 は IterDataPipes の正確に同じスタックを表します。DataPipe の関数型形式を使用することを推奨します。

以上