Dataset আর DataLoader — batching, shuffling আর বাংলা ডেটা

পর্ব ১০ (Training loop — loss, optimizer, train/eval mode আর checkpoint) প্রতিটি ধাপে “একটি batch” নেয়। কিন্তু batch আসে কোথা থেকে? ডেটা ফাইলে থাকে কাঁচা লেখা আর সংখ্যা হিসেবে; model চায় সমান আকারের tensor, এলোমেলো ক্রমে, প্রতি epoch-এ নতুন করে। এই পথটাই data pipeline. AI-এর ভিত্তি কোর্সের পর্ব ৬৭-এ বাংলা রিভিউ পরিষ্কার করা আর tokenization দেখেছেন; এখানে সেই কাজ PyTorch-এর কাঠামোয় বসবে। পাতার তিনটি ব্লক ব্রাউজারেই চলে (Python আর NumPy; প্রথমবার প্যাকেজ নামতে কয়েক সেকেন্ড লাগে)। PyTorch পাতায় চলে না, তাই বাকি সব ডেস্কটপের সত্যিকারের run: Python 3.14.4, PyTorch 2.15.0 (CPU build), torchvision 0.29.2, NumPy 2.5.4, লেখার সময়ের সংস্করণ।

১ · Dataset — “কয়টি আছে” আর “i-তম কোনটি”

PyTorch-এর সাধারণ Dataset-কে বলে map-style dataset: একটি class, যার দুটি method. __len__ মোট নমুনার সংখ্যা দেয়, __getitem__(i) i-তম নমুনা। কোনো জাদু নেই; Python কোর্সের পর্ব ৩২-এর dunder method দিয়েই len(ds) আর ds[i] কাজ করে। পুরো pipeline নিচের ছবিতে।

বাজারঘরের দুটি Dataset বানানো হলো, ফাইল bz_data.py-তে। OrdersDataset টেবিলের ডেটা: ১০০০টি অর্ডারের দূরত্ব আর ওজন থেকে ডেলিভারির ঘণ্টা (seed দেওয়া, বানানো)। পুরো ডেটা একবারেই float32 tensor-এ রাখা, __getitem__ শুধু একটি সারি কেটে দেয়। ReviewDataset বাংলা টেক্সট: make_reviews() ৬টি প্রশংসা আর ৫টি অভিযোগের বাক্যাংশ থেকে ১–৩টি জুড়ে ২০০টি রিভিউ বানায়, অভিযোগের সম্ভাবনা ১২%. ইচ্ছা করে অর্ধেক রিভিউতে “য়” এক code point-এ (U+09DF) লেখা, বাকিগুলোতে দুটিতে (য + নুক্তা), যেমন বাস্তবের কপি-পেস্ট করা ডেটায় হয়।

import unicodedata
from collections import Counter
import torch
from torch.utils.data import Dataset

def tokenize(text):
    return unicodedata.normalize("NFC", text).split()

def build_vocab(texts, min_freq=2):
    freq = Counter(w for t in texts for w in tokenize(t))
    itos = ["<pad>", "<unk>"] + sorted(w for w, c in freq.items() if c >= min_freq)
    return {w: i for i, w in enumerate(itos)}

class ReviewDataset(Dataset):
    def __init__(self, texts, labels, vocab):
        self.texts, self.labels, self.vocab = texts, labels, vocab
    def __len__(self):
        return len(self.texts)
    def __getitem__(self, i):
        ids = [self.vocab.get(w, 1) for w in tokenize(self.texts[i])]   # 1 = <unk>
        return torch.tensor(ids, dtype=torch.long), self.labels[i]

তিনটি নকশাগত সিদ্ধান্ত। এক, Unicode normalization tokenize-এর ভেতরে, তাই train, test আর পরে আসা নতুন রিভিউ সবাই একই পথে যায়। দুই, id ০ রাখা <pad>-এর জন্য, id ১ <unk>-এর জন্য; দুবারের কম দেখা শব্দ vocab-এ ঢোকে না। তিন, Dataset ভিন্ন দৈর্ঘ্যের tensor ফেরত দেয়; সমান করার দায়িত্ব পরে collate_fn-এর। ডেস্কটপে uv দিয়ে আলাদা project বানানো হলো (uv-এর পরিচয় AI-এর ভিত্তি কোর্সের পর্ব ৭-এ):

~/bazarghar-dl$ uv init --vcs none --author-from none --name bazarghar-dl --python 3.14
# pyproject.toml-এ PyTorch-এর CPU wheel index যোগ করা হলো (explicit = true)
~/bazarghar-dl$ uv add "torch==2.15.0" torchvision numpy
 …
 + torch==2.15.0+cpu
 + torchvision==0.29.2+cpu

datasets_demo.py দুটি Dataset বানায়, NFC-র প্রভাব গোনে, random_split দিয়ে ৭০:১৫:১৫ ভাগ করে আর শুধু train ভাগ থেকে vocab বানায়। সত্যিকারের run:

~/bazarghar-dl$ uv run python datasets_demo.py
Python 3.14.4 | torch 2.15.0+cpu | numpy 2.5.4
orders: 1000 | item 0: tensor([19.1278,  6.9786]) tensor([53.2907]) | torch.float32 (2,)
reviews: 200 | অভিযোগ: 23
আলাদা শব্দ: কাঁচা 28 | NFC-র পরে 26 | এক-code-point য়-ওয়ালা রিভিউ: 59
split: 140 30 30 | train-এর প্রথম ৫: [142, 56, 198, 77, 50]
vocab: 28 | [('<pad>', 0), ('<unk>', 1), ('অনুযায়ী', 2), ('অনেক', 3)]
train[0]: দাম অনুযায়ী মান ভালো পণ্য খুব ভালো প্যাকেট সুন্দর ছিল -> [14, 2, 24, 21, 17, 8, 21, 18, 27, 11] label 0
নতুন রিভিউ: [17, 1, 13, 1] (1 = <unk>)

টেবিলের একটি নমুনা (দূরত্ব ১৯.১ কিমি, ওজন ৭.০ কেজি) থেকে ঘণ্টা ৫৩.৩; dtype float32, আকার (2,). রিভিউতে ২৩টি অভিযোগ, মানে প্রায় ১১.৫%. “সময়মতো” আর “অনুযায়ী” দুই রূপে থাকায় কাঁচা ডেটায় ২৮টি আলাদা শব্দ, NFC-র পরে ২৬টি: NFC ছাড়া model এই দুটি শব্দকে চারটি আলাদা শব্দ ভাবত, আর প্রতিটির উদাহরণ অর্ধেক হয়ে যেত। Vocab-এ ২৬টি শব্দ আর দুটি বিশেষ token. শেষ লাইনে “চমৎকার” আর “দ্রুত” train-এ ছিল না, তাই দুটিই ১, অর্থাৎ <unk>.

২ · DataLoader — batch, shuffle আর shuffle-এর নিজের seed

DataLoader-এর মূল কাজ তিনটি: index সাজানো (shuffle=True হলে প্রতি epoch-এ নতুন এলোমেলো ক্রম), batch_size করে কাটা, আর প্রতিটি টুকরোকে collate করা। N-টি নমুনা আর batch size B হলে প্রতি epoch-এ batch সংখ্যা ⌈N/B⌉, শেষ batch-এ N mod B-টি নমুনা (০ হলে পূর্ণ)। drop_last=True দিলে শেষের ছোট batch বাদ, batch সংখ্যা ⌊N/B⌋. ছোট শেষ batch সমস্যা করে যখন কোনো স্তর batch-এর পরিসংখ্যান নেয় (পর্ব ১৪-এ BatchNorm), বা যখন সব batch-এর আকার একই হওয়া দরকার।

নিচের ব্লকে NumPy দিয়ে DataLoader-এর মূল যুক্তির একটি ছোট নকল। শেষ অংশে একটি পরীক্ষা: shuffle আর weight init যদি একই random উৎস ভাগ করে, তাহলে কী হয়।

import numpy as np

X = np.arange(10) * 10                   # ১০টি অর্ডার (মান: 0, 10, … 90)

def loader(n, batch_size, rng, shuffle=True, drop_last=False):
    """DataLoader-এর মূল কাজ: index সাজানো, তারপর batch_size করে কাটা।"""
    idx = rng.permutation(n) if shuffle else np.arange(n)
    for s in range(0, n, batch_size):
        b = idx[s:s + batch_size]
        if drop_last and len(b) < batch_size:
            break
        yield b

data_rng = np.random.default_rng(42)     # শুধু shuffle-এর জন্য আলাদা RNG
for epoch in range(2):
    print("epoch", epoch, [X[b].tolist() for b in loader(10, 4, data_rng)])

n, B = 10, 4
print("batch সংখ্যা: ceil =", -(-n // B), "| drop_last =", n // B)

# আরেকটি random কাজ (যেমন weight init) মাঝে ঢুকলে কী হয়?
def first_batch(share_rng, init_first):
    rng = np.random.default_rng(0)
    if init_first:
        rng.normal(size=5)               # model-এর ৫টি weight
    shuffle_rng = rng if share_rng else np.random.default_rng(1)
    return next(loader(10, 4, shuffle_rng)).tolist()

for share in (True, False):
    print("একই RNG" if share else "আলাদা RNG", "| init আগে নয়:", first_batch(share, False),
          "| init আগে:", first_batch(share, True))

epoch 0 [[50, 60, 0, 70], [30, 20, 40, 90], [10, 80]]
epoch 1 [[40, 80, 20, 60], [50, 90, 70, 30], [0, 10]]
batch সংখ্যা: ceil = 3 | drop_last = 2
একই RNG | init আগে নয়: [4, 6, 2, 7] | init আগে: [2, 8, 3, 6]
আলাদা RNG | init আগে নয়: [8, 4, 7, 0] | init আগে: [8, 4, 7, 0]

PyTorch-এ নিজের উৎস মানে একটি torch.Generator. loader_demo.py-র প্রথম অংশ:

import torch
from torch.utils.data import DataLoader, TensorDataset
from bz_data import ReviewDataset, make_reviews, build_vocab, pad_collate

ten = TensorDataset(torch.arange(10))                       # ১০টি সারি: 0..9
def orders(seed, epochs=2):
    dl = DataLoader(ten, batch_size=4, shuffle=True,
                    generator=torch.Generator().manual_seed(seed))
    return [[b[0].tolist() for b in dl] for _ in range(epochs)]
print("seed 42:", orders(42))
print("seed 42:", orders(42), "<- আবার চালালে")
print("seed  7:", orders(7))
dl = DataLoader(ten, batch_size=4, drop_last=True)
print("drop_last=True:", [b[0].tolist() for b in dl], "| len(dl) =", len(dl))

texts, labels = make_reviews()
ds = ReviewDataset(texts, labels, build_vocab(texts))
try:
    next(iter(DataLoader(ds, batch_size=3)))                 # default collate
except RuntimeError as e:
    print("RuntimeError:", e)
dl = DataLoader(ds, batch_size=3, shuffle=True, collate_fn=pad_collate,
                generator=torch.Generator().manual_seed(42))
ids, mask, lengths, y = next(iter(dl))
print("ids", tuple(ids.shape), ids.dtype, "| lengths", lengths.tolist(), "| y", y.tolist())
print(ids); print(mask.int())

সত্যিকারের run:

~/bazarghar-dl$ uv run python loader_demo.py
seed 42: [[[6, 5, 4, 0], [8, 9, 2, 1], [3, 7]], [[3, 6, 5, 8], [9, 4, 2, 0], [7, 1]]]
seed 42: [[[6, 5, 4, 0], [8, 9, 2, 1], [3, 7]], [[3, 6, 5, 8], [9, 4, 2, 0], [7, 1]]] <- আবার চালালে
seed  7: [[[1, 3, 5, 7], [9, 4, 6, 2], [8, 0]], [[5, 1, 6, 0], [9, 7, 8, 4], [3, 2]]]
drop_last=True: [[0, 1, 2, 3], [4, 5, 6, 7]] | len(dl) = 2
RuntimeError: stack expects each tensor to be equal size, but got [10] at entry 0 and [6] at entry 1
ids (3, 8) torch.int64 | lengths [8, 5, 6] | y [0, 0, 0]
tensor([[ 4,  7, 13, 26,  6, 25, 10, 23],
        [ 4,  7, 25, 10, 23,  0,  0,  0],
        [13, 26,  6, 17,  8, 21,  0,  0]])
tensor([[1, 1, 1, 1, 1, 1, 1, 1],
        [1, 1, 1, 1, 1, 0, 0, 0],
        [1, 1, 1, 1, 1, 1, 0, 0]], dtype=torch.int32)

একই Generator seed-এ দুই epoch-এর পুরো ক্রম হুবহু ফিরে এল, আর একই DataLoader-এর দ্বিতীয় epoch প্রথমটির থেকে আলাদা: Generator এগিয়ে চলে। seed ৭ সম্পূর্ণ অন্য ক্রম। drop_last-এ ৮ আর ৯ বাদ, len(dl) = ২. পঞ্চম লাইনের RuntimeError পরের অংশের বিষয়: default collate সব নমুনাকে torch.stack করতে চায়, আর দৈর্ঘ্য ১০ আর ৬ এক matrix-এ বসে না।

৩ · collate_fn — padding, lengths আর attention mask

Batch মানে একটি আয়তাকার tensor. ভিন্ন দৈর্ঘ্যের রিভিউকে আয়তাকার করার সহজ উপায় padding: সবচেয়ে লম্বা রিভিউয়ের দৈর্ঘ্য L_max ধরে বাকিগুলোর শেষে <pad> (id ০) বসানো। কিন্তু pad কোনো শব্দ নয়; model-কে জানাতে হবে কোন ঘর আসল। সেই খবর বহন করে attention mask: আকার ids-এর সমান, আসল token-এ ১ (True), pad-এ ০. নামটি এসেছে Transformer থেকে, যেখানে attention pad-এর দিকে না তাকাতে এটি লাগে (পর্ব ৩৩); কিন্তু গড় নেওয়া, loss হিসাব, RNN সবখানেই একই mask কাজে লাগে। collate_fn এই সব একসঙ্গে বানায়:

def pad_collate(batch, pad_id=0):
    seqs, labels = zip(*batch)                    # [(ids, y), …] থেকে দুই তালিকা
    lengths = torch.tensor([len(s) for s in seqs])
    ids = torch.full((len(seqs), int(lengths.max())), pad_id, dtype=torch.long)
    for r, s in enumerate(seqs):
        ids[r, :len(s)] = s
    mask = torch.arange(ids.shape[1])[None, :] < lengths[:, None]
    return ids, mask, lengths, torch.tensor(labels)

Mask-এর লাইনটি broadcasting: (1, L_max) আকারের 0, 1, …, L_max−1-কে (B, 1) আকারের lengths-এর সঙ্গে তুলনা করলে (B, L_max) আকারের True/False matrix. ids != pad_id লিখলেও প্রায়ই চলে, কিন্তু lengths থেকে বানানো mask নিরাপদ: কোনো tokenizer-এ আসল token-এর id ০ হলেও ভুল হয় না। PyTorch-এ torch.nn.utils.rnn.pad_sequence একই padding করে দেয়। নিচের stepper আগের run-এর সেই প্রথম batch ধাপে ধাপে দেখায়।

এবার একই কাজ NumPy-তে, তিনটি ছোট রিভিউ দিয়ে। শেষে দেখা যাবে mask ছাড়া গড় নিলে কী হয়: এখানে প্রতিটি token-এর একটি নকল “score” (id থেকে একটি সংখ্যা) ধরা হয়েছে, যেমন পরে embedding-এর গড় নেওয়া হবে।

import numpy as np

vocab = {"<pad>": 0, "<unk>": 1, "অনেক": 3, "আবার": 4, "এসেছে": 6, "কিনব": 7,
         "চাই": 9, "টাকা": 12, "ডেলিভারি": 13, "দেরিতে": 15, "পণ্য": 17,
         "ফেরত": 19, "ভাঙা": 20}
reviews = ["আবার কিনব", "পণ্য ভাঙা এসেছে", "ডেলিভারি অনেক দেরিতে টাকা ফেরত চাই"]
seqs = [[vocab.get(w, 1) for w in r.split()] for r in reviews]
print("token id:", seqs)

def pad_collate(seqs, pad_id=0):
    lengths = np.array([len(s) for s in seqs])
    ids = np.full((len(seqs), lengths.max()), pad_id, dtype=np.int64)
    for r, s in enumerate(seqs):
        ids[r, :len(s)] = s
    mask = np.arange(ids.shape[1])[None, :] < lengths[:, None]   # broadcasting
    return ids, mask, lengths

ids, mask, lengths = pad_collate(seqs)
print("ids", ids.shape); print(ids)
print("mask"); print(mask.astype(int)); print("lengths", lengths)
print(f"pad ঘর: {(~mask).sum()} / {mask.size} = {(~mask).mean():.0%}")

# প্রতিটি token-এর একটি নকল score; গড় নেওয়ার দুই উপায়
emb = np.linspace(-1, 1, 21)             # id -> একটি সংখ্যা
score = emb[ids]                         # (3, 6)
naive = score.mean(axis=1)               # pad-ও গড়ে ঢুকছে
masked = (score * mask).sum(axis=1) / mask.sum(axis=1)
print("naive mean :", naive.round(3))
print("masked mean:", masked.round(3))

token id: [[4, 7], [17, 20, 6], [13, 3, 15, 12, 19, 9]]
ids (3, 6)
[[ 4  7  0  0  0  0]
 [17 20  6  0  0  0]
 [13  3 15 12 19  9]]
mask
[[1 1 0 0 0 0]
 [1 1 1 0 0 0]
 [1 1 1 1 1 1]]
lengths [2 3 6]
pad ঘর: 7 / 18 = 39%
naive mean : [-0.817 -0.283  0.183]
masked mean: [-0.45   0.433  0.183]

Sequence ডেটার padding, masking আর RNN-এর জন্য packing বিস্তারিত আসবে পর্ব ২৭-এ (Sequence ডেটা — সময়, টেক্সট আর padding ও masking). Bucketing-এর ধারণা এখানেই মনে রাখুন: দৈর্ঘ্যে কাছাকাছি নমুনা একই batch-এ রাখলে pad কমে, কিন্তু shuffle কম এলোমেলো হয়, তাই সাধারণত bucket-এর ভেতরে আর bucket-গুলোর ক্রমে আলাদা করে shuffle করা হয়।

৪ · Sampler — বিরল শ্রেণিকে বেশিবার দেখানো

২০০টি রিভিউয়ের ২৩টি অভিযোগ: এটি class imbalance. সাধারণ shuffle-এ ৩২-এর একটি batch-এ গড়ে মাত্র ৩–৪টি অভিযোগ থাকে, আর model “সব রিভিউ ভালো” বলেও প্রায় ৮৮% ঠিক পায়। মেশিন লার্নিং কোর্সের পর্ব ৫০-এ oversampling, class weight আর threshold-এর তুলনা হয়েছে। Deep learning-এ oversampling-এর সবচেয়ে সহজ জায়গা sampler: DataLoader-কে কোন index দেওয়া হবে, তা sampler ঠিক করে। WeightedRandomSampler প্রতিটি সারিকে একটি ওজন দেয় আর সেই অনুপাতে replacement-সহ index টানে।

হিসাব: শ্রেণি c-তে n_c-টি সারি থাকলে প্রতিটি সারির ওজন w = 1/n_c. তাহলে শ্রেণি c-এর মোট ওজন n_c × 1/n_c = ১, সব শ্রেণির সমান; সম্ভাবনায় প্রতিটি শ্রেণি 1/K (K = শ্রেণির সংখ্যা)। একটি সারি টানার সম্ভাবনা p = w / Σw. N বার টানলে একটি নির্দিষ্ট সারি অন্তত একবার আসার সম্ভাবনা 1 − (1 − p)^N.

import numpy as np

y = np.array([0] * 177 + [1] * 23)       # ২০০টি রিভিউ, ২৩টি অভিযোগ
counts = np.bincount(y)
w_class = 1 / counts                     # শ্রেণির ওজন = 1 / সংখ্যা
w_row = w_class[y]                       # প্রতিটি সারির ওজন
p = w_row / w_row.sum()                  # টানার সম্ভাবনা
print("counts:", counts, "| p (সংখ্যাগুরু সারি):", round(float(p[0]), 5),
      "| p (অভিযোগ সারি):", round(float(p[-1]), 5))
print("প্রত্যাশিত শ্রেণি-ভাগ:", [round(float(p[y == c].sum()), 3) for c in (0, 1)])

rng = np.random.default_rng(0)
n = len(y)
for epoch in range(3):
    idx = rng.choice(n, size=n, replace=True, p=p)
    print(f"epoch {epoch}: শ্রেণি {np.bincount(y[idx], minlength=2)}",
          f"| আলাদা সারি {len(np.unique(idx))}",
          f"| অভিযোগ সারির গড় পুনরাবৃত্তি {np.mean(y[idx]) * n / 23:.2f}")

# একটি নির্দিষ্ট সারি এক epoch-এ অন্তত একবার আসার সম্ভাবনা: 1 - (1 - p)^n
for c in (0, 1):
    pr = 1 - (1 - p[y == c][0]) ** n
    print(f"শ্রেণি {c}: অন্তত একবার আসে {pr:.3f}")
print("প্রত্যাশিত আলাদা সারি:", round(float(177 * (1 - (1 - p[0]) ** n) + 23 * (1 - (1 - p[-1]) ** n)), 1))

counts: [177  23] | p (সংখ্যাগুরু সারি): 0.00282 | p (অভিযোগ সারি): 0.02174
প্রত্যাশিত শ্রেণি-ভাগ: [0.5, 0.5]
epoch 0: শ্রেণি [ 88 112] | আলাদা সারি 95 | অভিযোগ সারির গড় পুনরাবৃত্তি 4.87
epoch 1: শ্রেণি [ 95 105] | আলাদা সারি 93 | অভিযোগ সারির গড় পুনরাবৃত্তি 4.57
epoch 2: শ্রেণি [ 91 109] | আলাদা সারি 94 | অভিযোগ সারির গড় পুনরাবৃত্তি 4.74
শ্রেণি 0: অন্তত একবার আসে 0.432
শ্রেণি 1: অন্তত একবার আসে 0.988
প্রত্যাশিত আলাদা সারি: 99.2

এবার PyTorch-এ, ReviewDataset আর pad_collate-সহ (sampler_demo.py-র মূল অংশ):

y = torch.tensor(labels)
counts = torch.bincount(y)
w_class = 1.0 / counts.float()                    # বিরল শ্রেণির ওজন বেশি
weights = w_class[y]                              # প্রতিটি সারির ওজন
plain = DataLoader(ds, batch_size=32, shuffle=True, collate_fn=pad_collate,
                   generator=torch.Generator().manual_seed(0))
sampler = WeightedRandomSampler(weights, num_samples=len(ds), replacement=True,
                                generator=torch.Generator().manual_seed(0))
bal = DataLoader(ds, batch_size=32, sampler=sampler, collate_fn=pad_collate)
for ep in range(3):
    print(f"epoch {ep}: shuffle {epoch_counts(plain)} | weighted {epoch_counts(bal)}")
idx = list(sampler)                               # আরেকবার (চতুর্থ) টানা
rare = [i for i in idx if labels[i] == 1]
print("চতুর্থ টানায় বিরল শ্রেণি:", len(rare), "| আলাদা সারি:", len(set(rare)),
      "| মোট আলাদা সারি:", len(set(idx)), "/", len(ds))

~/bazarghar-dl$ uv run python sampler_demo.py
শ্রেণির সংখ্যা: [177, 23]
প্রতি শ্রেণির ওজন: [0.0056, 0.0435]
epoch 0: shuffle [177, 23] | weighted [100, 100]
epoch 1: shuffle [177, 23] | weighted [101, 99]
epoch 2: shuffle [177, 23] | weighted [99, 101]
চতুর্থ টানায় বিরল শ্রেণি: 98 | আলাদা সারি: 23 | মোট আলাদা সারি: 98 / 200

Shuffle প্রতি epoch-এ ঠিক ১৭৭:২৩ দেয়, কারণ সে শুধু ক্রম বদলায়। Weighted sampler প্রায় ১০০:১০০, তবে টানা random বলে প্রতিবার একটু ওঠানামা। শেষ লাইনটিই দাম: ২৩টি অভিযোগ ৯৮ বার এসেছে, প্রতিটি গড়ে চারবারের বেশি, আর ২০০ টানায় মাত্র ৯৮টি আলাদা সারি; সংখ্যাগুরু শ্রেণির অর্ধেকের বেশি সারি এই epoch-এ দেখাই হয়নি। দুটি সতর্কতা: sampler আর shuffle=True একসঙ্গে দেওয়া যায় না (ValueError: “sampler option is mutually exclusive with shuffle”), আর validation বা test-এ কখনো weighted sampler নয়; সেখানে আসল অনুপাতেই মাপতে হয়।

৫ · ভাগ করা — random_split, আগে ভাগ পরে Dataset

অংশ ১-এর run-এ random_split(range(200), [0.7, 0.15, 0.15], generator=g) ১৪০, ৩০, ৩০-এ ভাগ করেছে। ভগ্নাংশ দিলে PyTorch সংখ্যায় রূপান্তর করে নেয়; Generator ছাড়া প্রতিবার আলাদা ভাগ আসত, আর আজকের test সেট কালকের train-এ ঢুকে যেত। দুটি প্রচলিত পথ আছে:

আমরা দ্বিতীয় পথ নিয়েছি। ৩০টির test সেটে অভিযোগ মাত্র কয়েকটি থাকতে পারে, তাই বাস্তব কাজে stratified ভাগ ভালো; আর একই গ্রাহক বা প্রায় একই লেখা দুই ভাগে পড়লে নকল-ভালো ফল আসে। এখানে অনেক রিভিউ একই বাক্যাংশ থেকে তৈরি (যেমন অংশ ১-এর ১৪২ আর ০ নম্বর হুবহু এক লেখা), তাই এই বানানো ডেটায় test-এর ফল আশাবাদী হবে; বাস্তব ডেটায় ভাগ করার আগে নকল সারি খোঁজা জরুরি।

৬ · ছবির transforms.v2 আর IterableDataset — এক ঝলক

ছবির Dataset-এর __getitem__ সাধারণত ফাইল পড়ে, তারপর torchvision.transforms.v2-এর একটি pipeline চালায়: HWC uint8 array থেকে CHW tensor, random crop আর flip (training-এ data augmentation), float-এ রূপান্তর, normalisation. বাজারঘরের বদলে এখানে ১২টি ছোট কৃত্রিম “পণ্যের ছবি” (random pixel, আকার ৪০ × ৪৮). ছবি যে array, তা AI-এর ভিত্তি কোর্সের পর্ব ১৮-এ দেখেছেন; augmentation-এর পুরো আলোচনা পর্ব ২৪-এ।

import numpy as np, torch, torchvision
from torchvision.transforms import v2
from torch.utils.data import Dataset, DataLoader
print("torchvision", torchvision.__version__)

rng = np.random.default_rng(3)                       # ছোট কৃত্রিম "পণ্যের ছবি"
imgs = rng.integers(0, 256, size=(12, 40, 48, 3), dtype=np.uint8)   # N, H, W, C
labels = rng.integers(0, 3, size=12)

train_tf = v2.Compose([
    v2.ToImage(),                                    # HWC uint8 -> CHW tensor
    v2.RandomResizedCrop((32, 32), scale=(0.6, 1.0)),
    v2.RandomHorizontalFlip(p=0.5),
    v2.ToDtype(torch.float32, scale=True),           # 0..255 -> 0..1
    v2.Normalize(mean=[0.5] * 3, std=[0.25] * 3),
])

class ProductImages(Dataset):
    def __init__(self, imgs, labels, tf): self.imgs, self.labels, self.tf = imgs, labels, tf
    def __len__(self): return len(self.imgs)
    def __getitem__(self, i): return self.tf(self.imgs[i]), int(self.labels[i])

torch.manual_seed(0)                                 # v2-র random কাজ global RNG নেয়
ds = ProductImages(imgs, labels, train_tf)
x, y = ds[0]
print("আগে :", imgs[0].shape, imgs[0].dtype, imgs[0].min(), imgs[0].max())
print("পরে :", tuple(x.shape), x.dtype, round(x.min().item(), 3), round(x.max().item(), 3))
a, b = ds[0][0], ds[0][0]
print("একই ছবি দুবার পড়লে একই?", torch.equal(a, b))
xb, yb = next(iter(DataLoader(ds, batch_size=4)))
print("batch:", tuple(xb.shape), xb.dtype, "| labels", yb.tolist())

~/bazarghar-dl$ uv run python images_demo.py
torchvision 0.29.2+cpu
আগে : (40, 48, 3) uint8 0 255
পরে : (3, 32, 32) torch.float32 -1.639 1.702
একই ছবি দুবার পড়লে একই? False
batch: (4, 3, 32, 32) torch.float32 | labels [0, 0, 0, 1]

আকার (40, 48, 3) uint8 থেকে (3, 32, 32) float32: channel সামনে এল, crop ৩২ × ৩২-এ নামাল। (x/255 − 0.5)/0.25-এর তাত্ত্বিক সীমা −২ থেকে ২, কিন্তু resize পাশের pixel মিশিয়ে চরম মান কমায়, তাই −১.৬৪ থেকে ১.৭০. একই ছবি দুবার পড়লে আলাদা ফল: augmentation প্রতিটি __getitem__-এ নতুন করে random, তাই প্রতি epoch-এ model একটু ভিন্ন ছবি দেখে। এজন্য validation-এর transform আলাদা রাখতে হয়, random অংশ ছাড়া। Default collate এখানে চলেছে, কারণ সব ছবি এখন একই আকারের।

IterableDataset — যখন index নেই, শুধু stream

ডেটা যদি একটি বিশাল ফাইল বা চলমান উৎস হয়, যেখানে “১৪২ নম্বরটি দিন” বলা যায় না, তখন IterableDataset: __getitem__-এর বদলে __iter__, যা একটির পর একটি নমুনা yield করে (Python কোর্সের পর্ব ৩৬-এর generator). দুটি ফাঁদ নিচের run-এ:

import torch
from torch.utils.data import IterableDataset, DataLoader, get_worker_info

class ReviewStream(IterableDataset):          # যেমন: বড় ফাইল বা নেটওয়ার্ক থেকে লাইন ধরে
    def __init__(self, n, shard=True): self.n, self.shard = n, shard
    def __iter__(self):
        info = get_worker_info()
        start, step = 0, 1
        if self.shard and info is not None:   # প্রতিটি worker আলাদা ভাগ পড়ে
            start, step = info.id, info.num_workers
        for i in range(start, self.n, step):
            yield torch.tensor(i)

if __name__ == "__main__":
    for shard in (False, True):
        dl = DataLoader(ReviewStream(8, shard), batch_size=4, num_workers=2)
        print(f"shard={shard}:", [b.tolist() for b in dl])
    try:
        DataLoader(ReviewStream(8), batch_size=4, shuffle=True)
    except ValueError as e:
        print("ValueError:", e)

~/bazarghar-dl$ uv run python stream_demo.py
shard=False: [[0, 1, 2, 3], [0, 1, 2, 3], [4, 5, 6, 7], [4, 5, 6, 7]]
shard=True: [[0, 2, 4, 6], [1, 3, 5, 7]]
ValueError: DataLoader with IterableDataset: expected unspecified shuffle option, but got shuffle=True

দুটি worker-এর প্রতিটি Dataset-এর নিজের কপি পায়; ভাগ না করলে দুজনেই পুরো stream পড়ে, আর প্রতিটি নমুনা দুবার আসে, কোনো error ছাড়াই। get_worker_info() দিয়ে worker নিজের ভাগ বেছে নিলে প্রতিটি একবার। আর IterableDataset-এ shuffle=True চলে না; stream-এ এলোমেলো করতে সাধারণত একটি ছোট buffer-এ জমিয়ে তার ভেতরে মেশানো হয়, যা পুরো shuffle নয়।

৭ · num_workers, Windows-এর __main__ পাহারা আর মাপা গতি

num_workers=0 হলে মূল process নিজেই প্রতিটি নমুনা আনে; model-এর হিসাব আর ডেটা আনা পালা করে চলে। num_workers=k হলে k-টি আলাদা worker process আগেভাগে batch বানিয়ে রাখে। Linux-এ বহুদিন default ছিল fork (মূল process-এর কপি, দ্রুত; Python 3.14 থেকে Linux-এর default forkserver); Windows-এ (আর macOS-এও default) spawn: প্রতিটি worker একটি নতুন Python interpreter, যা আপনার মূল script আবার import করে। Script-এর একদম ওপরে DataLoader চালানোর কোড থাকলে প্রতিটি worker নিজেও worker বানাতে চায়। পাহারা ছাড়া একটি script চালানোর সত্যিকারের ফল (path আর pid ছাঁটা):

~/bazarghar-dl$ uv run python noguard.py      # num_workers=2, if __name__ নেই
Traceback (most recent call last):
  …
RuntimeError:
        An attempt has been made to start a new process before the
        current process has finished its bootstrapping phase.
        …
            if __name__ == '__main__':
                freeze_support()
  …
RuntimeError: DataLoader worker (pid(s) …) exited unexpectedly

সমাধান: DataLoader তৈরি আর লুপ if __name__ == "__main__":-এর ভেতরে, আর Dataset class আর collate_fn module-এর ওপরের স্তরে (worker যেন import করে পায়; lambda বা ভেতরের function spawn-এ pickle হয় না)। Process আর thread-এর পার্থক্য Python কোর্সের পর্ব ৪৪-এ। Jupyter notebook-এ Windows-এ worker প্রায়ই এ কারণেই আটকায়; তখন Dataset একটি .py ফাইলে রেখে import করুন, বা num_workers=0.

এবার মাপা যাক worker আসলে কতটা সাহায্য করে। throughput.py-তে ৬৪০০টি নমুনা, batch ৬৪, দুই epoch; “হালকা” Dataset প্রায় কিছুই করে না, “ভারী” প্রতি নমুনায় একটি Python লুপ চালায় (টেক্সট পরিষ্কার আর tokenize-এর খরচের নকল)। প্রতিটি epoch-এ প্রথম batch আসার সময় আর মোট সময় মাপা হলো; persistent_workers=True হলে worker epoch-এর শেষে বন্ধ হয় না।

import time, torch
from torch.utils.data import Dataset, DataLoader

class Reviews(Dataset):
    def __init__(self, n, work): self.n, self.work = n, work
    def __len__(self): return self.n
    def __getitem__(self, i):
        x = torch.full((64,), float(i % 97))
        s = 0
        for k in range(self.work):            # প্রতি সারিতে নকল "পরিষ্কার আর tokenize" খরচ
            s += (k * i) % 7
        return x + s, i % 2

def run(ds, nw, epochs=2, persistent=False):
    dl = DataLoader(ds, batch_size=64, shuffle=True, num_workers=nw,
                    persistent_workers=persistent and nw > 0,
                    generator=torch.Generator().manual_seed(0))
    out = []
    for _ in range(epochs):
        t0 = time.perf_counter(); first = None
        for b, (x, y) in enumerate(dl):
            if b == 0: first = time.perf_counter() - t0
        out.append((first, time.perf_counter() - t0))
    return out

if __name__ == "__main__":                    # Windows-এ worker spawn-এর জন্য বাধ্যতামূলক
    print("torch", torch.__version__)
    for name, work in [("হালকা", 0), ("ভারী", 20000)]:
        ds = Reviews(6400, work)
        for nw in (0, 2, 4):
            for pers in ((False, True) if nw else (False,)):
                (f1, t1), (f2, t2) = run(ds, nw, persistent=pers)
                tag = f"{name} nw={nw}" + (" persistent" if pers else "")
                print(f"{tag:<22} epoch1 প্রথম batch {f1:6.3f}s মোট {t1:6.2f}s"
                      f" | epoch2 প্রথম batch {f2:6.3f}s মোট {t2:6.2f}s")

সত্যিকারের run (ডেস্কটপ, Windows, CPU-তে; একটি run-এর মাপ, অন্য যন্ত্রে সংখ্যা আলাদা হবে):

~/bazarghar-dl$ uv run python throughput.py
torch 2.15.0+cpu
হালকা nw=0             epoch1 প্রথম batch  0.001s মোট   0.03s | epoch2 প্রথম batch  0.000s মোট   0.04s
হালকা nw=2             epoch1 প্রথম batch  1.342s মোট   1.66s | epoch2 প্রথম batch  1.336s মোট   1.66s
হালকা nw=2 persistent  epoch1 প্রথম batch  1.374s মোট   1.41s | epoch2 প্রথম batch  0.001s মোট   0.03s
হালকা nw=4             epoch1 প্রথম batch  1.375s মোট   1.74s | epoch2 প্রথম batch  1.469s মোট   1.81s
হালকা nw=4 persistent  epoch1 প্রথম batch  1.415s মোট   1.44s | epoch2 প্রথম batch  0.002s মোট   0.02s
ভারী nw=0              epoch1 প্রথম batch  0.041s মোট   4.10s | epoch2 প্রথম batch  0.040s মোট   4.08s
ভারী nw=2              epoch1 প্রথম batch  1.350s মোট   3.63s | epoch2 প্রথম batch  1.338s মোট   3.61s
ভারী nw=2 persistent   epoch1 প্রথম batch  1.345s মোট   3.34s | epoch2 প্রথম batch  0.042s মোট   2.13s
ভারী nw=4              epoch1 প্রথম batch  1.460s মোট   2.77s | epoch2 প্রথম batch  1.383s মোট   2.71s
ভারী nw=4 persistent   epoch1 প্রথম batch  1.403s মোট   2.41s | epoch2 প্রথম batch  0.056s মোট   1.09s

এই সংখ্যা থেকে তিনটি শিক্ষা। এক, worker চালু হওয়ার দাম: প্রতিটি epoch-এ worker নতুন করে spawn হলে প্রথম batch আসতে প্রায় ১.৩–১.৫ সেকেন্ড, কারণ প্রতিটি worker-এ Python আর torch আবার import হয়। দুই, হালকা ডেটায় worker ক্ষতি করে: ০.০৩ সেকেন্ডের কাজ ১.৭ সেকেন্ডে উঠল; টেবিলের মতো আগে থেকে tensor-এ রাখা ডেটায় num_workers=0-ই সবচেয়ে দ্রুত। তিন, ভারী ডেটায় worker সাহায্য করে, persistent হলে আরও: দ্বিতীয় epoch-এ ৪.০৮ থেকে ১.০৯ সেকেন্ড (চারটি persistent worker), প্রায় ৩.৭ গুণ throughput; non-persistent চারটি worker-এ ২.৭১, কারণ প্রতি epoch-এ আবার spawn-এর দাম। Linux-এ (fork বা forkserver) worker চালুর দাম সাধারণত কম বলে জানা যায়, তাই এই সংখ্যা শুধু এই Windows যন্ত্রের একটি run.

pin_memory=True batch-কে page-locked memory-তে রাখে, যাতে GPU-তে কপি দ্রুত আর non_blocking=True হয়। এই CPU build-এ চালালে PyTorch সতর্ক করে: UserWarning: 'pin_memory' argument is set as true but no accelerator is found, then device pinned memory won't be used. GPU-তে ডেটা পাঠানো আর ডেটা-লোডিং-এর bottleneck ধরার profiler পর্ব ১২-এ।

৮ · AI-এর যুগে — AI-এর লেখা data pipeline যাচাই

“আমার CSV থেকে একটি PyTorch DataLoader বানিয়ে দাও” বললে AI assistant সেকেন্ডে কোড দেয়, আর তা প্রায়ই চলে। কিন্তু data pipeline-এর বেশির ভাগ বাগ error দেয় না: ভুল vocab, pad-সহ গড়, দুবার পড়া stream, test-এ weighted sampler. Model শুধু একটু খারাপ শেখে, বা মাপা ফল একটু বেশি ভালো আসে।

AI-এর দেওয়া data pipeline হাতে পেলে — যাচাইয়ের চেকলিস্ট

• একটি batch বের করে প্রতিটি tensor-এর আকার, dtype আর mask-এর যোগফল = lengths-এর যোগফল কি না দেখেছি
• কয়েকটি id আবার শব্দে ফিরিয়ে আসল টেক্সটের সঙ্গে মিলিয়েছি, আর <unk>-এর হার গুনেছি
• ভাগ আগে, vocab বা scaler পরে, আর val/test-এ কোনো weighted sampler বা random augmentation নেই
• একই seed-এ দুবার চালিয়ে একই ক্রম পেয়েছি; model-এর কোড বদলালেও ডেটার ক্রম বদলায় না
• এক batch overfit করিয়ে (পর্ব ১৮-এর অভ্যাস) দেখেছি pipeline-এর ভুল না model-এর
• num_workers মেপে বেছেছি, অনুমান করে নয়; আর কোনো আসল গ্রাহকের রিভিউ বা key AI-তে পেস্ট করিনি