فصل ۱۳: بارگذاری و پیش‌پردازش داده با TensorFlow

صفحهٔ ۳۸ فایل PDF صفحهٔ جداکنندهٔ سفید میان دو فصل است و پس از آن فصل ۱۳ آغاز می‌شود. در پروژه‌های کوچک، Pandas و Transformerهای Scikit-Learn برای Load و Preprocess داده بسیار مناسب‌اند. اما وقتی TensorFlow Model روی Dataset بزرگ Train می‌شود، API اختصاصی tf.data می‌تواند داده را از چند File به‌صورت موازی بخواند، Shuffle و Batch کند و هم‌زمان با Training، Batch بعدی را روی CPU آماده سازد. این API Datasetهایی را که در RAM جا نمی‌شوند نیز Stream می‌کند و برای استفادهٔ کامل از CPU/GPU/TPU طراحی شده است.

فرمت‌های ورودی پشتیبانی‌شده شامل Text/CSV، Binary Record با اندازهٔ ثابت، TFRecord با Recordهای متغیر و حتی SQL Database است. Keras نیز Preprocessing Layerهایی دارد که می‌توان آن‌ها را داخل خود Model قرار داد؛ این کار خطر training/serving skew را کاهش می‌دهد، چون دقیقاً همان Preprocessing در Training و Production اجرا می‌شود.

API ‏tf.data

مفهوم مرکزی، tf.data.Dataset است: Sequenceای از Itemها که معمولاً به‌صورت Stream خوانده می‌شوند. ساده‌ترین مثال، ساخت Dataset از Tensor است:

import tensorflow as tf

X = tf.range(10)
dataset = tf.data.Dataset.from_tensor_slices(X)

for item in dataset:
    print(item)

from_tensor_slices() Tensor را در Dimension اول Slice می‌کند، بنابراین Dataset بالا شامل Tensorهای ۰ تا ۹ است. tf.data برای Iterate بسیار بهینه است اما برای Indexing و Slicing تصادفی طراحی نشده است.

Item هر Dataset می‌تواند Tensor، Tuple، Dictionary یا ساختار Nested از این‌ها باشد. هنگام Slice کردن ساختار Nested، Tensorهای داخلی Slice می‌شوند و شکل Dictionary/Tuple حفظ می‌شود.

زنجیره کردن Transformationها

متدهای Transformation Dataset اصلی را تغییر نمی‌دهند؛ هرکدام Dataset جدیدی می‌سازند. بنابراین باید Reference خروجی را نگه داشت:

dataset = tf.data.Dataset.from_tensor_slices(tf.range(10))
dataset = dataset.repeat(3).batch(7)

for item in dataset:
    print(item)

repeat(3) Sequence را سه بار تکرار می‌کند بدون اینکه سه Copy کامل در RAM بسازد. اگر بدون Argument فراخوانی شود Dataset تا بی‌نهایت تکرار می‌شود و Consumer باید شرط توقف داشته باشد. batch(7) Itemها را در Batchهای هفت‌تایی گروه‌بندی می‌کند؛ Batch آخر ممکن است کوچک‌تر باشد. برای حذف Batch ناقص از drop_remainder=True استفاده می‌شود.

با map() می‌توان هر Item را Transform کرد:

dataset = dataset.map(lambda x: x * 2)

برای Preprocessing سنگین، num_parallel_calls=tf.data.AUTOTUNE به TensorFlow اجازه می‌دهد تعداد Threadها را بر اساس منابع موجود تنظیم کند. Function ارسال‌شده به map() باید قابل تبدیل به TF Function باشد.

filter() Itemها را بر اساس Predicate نگه می‌دارد و take(n) برای مشاهدهٔ چند Item اول مفید است:

dataset = dataset.filter(lambda x: tf.reduce_sum(x) > 50)
for item in dataset.take(2):
    print(item)

Shuffle کردن داده

Gradient Descent معمولاً وقتی بهتر عمل می‌کند که Sampleهای Training تقریباً IID باشند. shuffle() یک Buffer می‌سازد، Itemی تصادفی از آن بیرون می‌کشد و با Item تازه از Source جایگزین می‌کند. Buffer باید به‌اندازهٔ کافی بزرگ باشد تا Shuffle مؤثر شود، ولی نباید RAM را بی‌دلیل مصرف کند:

dataset = tf.data.Dataset.range(10).repeat(2)
dataset = dataset.shuffle(buffer_size=4, seed=42).batch(7)

به‌طور پیش‌فرض اگر Dataset Shuffleشده Repeat شود، Order در هر Iteration جدید عوض می‌شود. برای Debug/Test می‌توان reshuffle_each_iteration=False تعیین کرد.

برای Dataset بسیار بزرگ، Buffer محدود به‌تنهایی Shuffle کامل ایجاد نمی‌کند. روش بهتر این است که Source Data نیز از قبل Shuffle شود، میان چند File تقسیم شود، Fileها به ترتیب Random خوانده شوند و Recordهای چند File به‌صورت Interleave وارد Pipeline شوند؛ سپس یک Shuffle Buffer نهایی نیز اعمال شود.

Interleave کردن چند File

فرض کنید California Housing به چند CSV File برای Train/Validation/Test تقسیم شده است. ابتدا Dataset مسیر Fileها را می‌سازیم:

filepath_dataset = tf.data.Dataset.list_files(
    train_filepaths, seed=42)

list_files() به‌طور پیش‌فرض مسیرها را Shuffle می‌کند. سپس پنج File را هم‌زمان باز و Lineها را Interleave می‌کنیم و Header هر File را رد می‌کنیم:

n_readers = 5
dataset = filepath_dataset.interleave(
    lambda filepath: tf.data.TextLineDataset(filepath).skip(1),
    cycle_length=n_readers,
    num_parallel_calls=tf.data.AUTOTUNE)

Interleave Dataset در هر لحظه چند TextLineDataset داخلی دارد و به‌صورت چرخه‌ای از آن‌ها Line می‌خواند. اگر Fileها طول تقریباً یکسان داشته باشند Interleaving مؤثرتر است. num_parallel_calls خواندن واقعاً موازی را فعال می‌کند.

Parse و Scale کردن CSV

Line خام CSV یک Tensor از Byte String است و باید Parse و Scale شود. مثال کتاب Mean و Standard Deviation هشت Feature را از Training Set از قبل محاسبه می‌کند:

X_mean, X_std = [...]  # یک مقدار برای هر feature
n_inputs = 8

def parse_csv_line(line):
    defs = [0.] * n_inputs + [tf.constant([], dtype=tf.float32)]
    fields = tf.io.decode_csv(line, record_defaults=defs)
    return tf.stack(fields[:-1]), tf.stack(fields[-1:])

def preprocess(line):
    x, y = parse_csv_line(line)
    return (x - X_mean) / X_std, y

record_defaults هم تعداد Columnها و Type آن‌ها را مشخص می‌کند و هم Default Value را. برای Target از Tensor خالی Float32 استفاده شده است؛ یعنی اگر Target گم‌شده باشد Exception ایجاد شود. tf.stack() Scalarهای خروجی Decoder را به Vector Feature و Vector تک‌عضوی Target تبدیل می‌کند.

ساخت Pipeline قابل‌استفادهٔ مجدد

def csv_reader_dataset(filepaths, n_readers=5, n_read_threads=None,
                       n_parse_threads=5, shuffle_buffer_size=10_000,
                       seed=42, batch_size=32):
    dataset = tf.data.Dataset.list_files(filepaths, seed=seed)
    dataset = dataset.interleave(
        lambda filepath: tf.data.TextLineDataset(filepath).skip(1),
        cycle_length=n_readers,
        num_parallel_calls=n_read_threads)
    dataset = dataset.map(preprocess,
                          num_parallel_calls=n_parse_threads)
    dataset = dataset.shuffle(shuffle_buffer_size, seed=seed)
    return dataset.batch(batch_size).prefetch(1)

در انتهای Pipeline از prefetch(1) استفاده شده است. این بخش در مقالهٔ بعدی توضیح داده می‌شود و برای هم‌پوشانی آماده‌سازی Data روی CPU با Training روی GPU اهمیت زیادی دارد.