বড় ডেটায় ML — out-of-core, incremental learning আর Polars/DuckDB

এই মডিউলের অন্য পর্বগুলো model-কে কাজে নামায়: পর্ব ৬৮ (Model serving — FastAPI, batch scoring আর ONNX) আর পর্ব ৬৯ (ML system-এর test আর নজরদারি — data test, model test আর drift). এই পর্বের প্রশ্ন আলাদা: ডেটা যখন ল্যাপটপের RAM-এ ধরে না, তখন কী করবেন? pandas-এর dtype, category, chunk আর Polars-এর প্রথম পরিচয় AI-এর ভিত্তি কোর্সের পর্ব ২৭-এ আছে, Parquet পড়া-লেখা একই কোর্সের পর্ব ২১-এ। এখানে সেগুলো ধরে নিয়ে আরও গভীরে যাব। চারটি চালানো যায় এমন ব্লক পাতাতেই চলে (ব্রাউজারের Python); প্রথমবার scikit-learn, Polars আর DuckDB নামতে প্রতিটিতে দশ সেকেন্ডের মতো লাগতে পারে। ১ কোটি সারির মাপ ডেস্কটপে সত্যিকারের run.

১ · ডেটা কখন “বড়” — RAM, কাজের memory আর চারটি পথ

“বড় ডেটা” কোনো নির্দিষ্ট সংখ্যা নয়; এটি আপনার যন্ত্রের সঙ্গে তুলনা। ১০ GB-এর CSV একটি ৬৪ GB RAM-এর server-এ সাধারণ ডেটা, কিন্তু ৮ GB-এর ল্যাপটপে বড়। আর ফাইলের আকার RAM-এর চাহিদার সমান নয়, তিনটি কারণে। (ক) CSV পড়ার পর আকার বদলায়: সংখ্যা ছোট হতে পারে, লেখার কলাম অনেক বড়। (খ) pandas-এর অনেক কাজ (filter, merge, নতুন কলাম) নতুন copy বানায়, তাই কাজের সময় memory টেবিলের কয়েক গুণ। pandas-এর স্রষ্টা Wes McKinney ২০১৭ সালের একটি লেখায় মোটামুটি নিয়ম দিয়েছিলেন: ডেটার আকারের ৫ থেকে ১০ গুণ RAM রাখুন। এটি অভিজ্ঞতার নিয়ম, মাপা সূত্র নয়। (গ) model শেখানোর সময় feature matrix আর তার ভেতরের copy আরও জায়গা নেয়।

ডেটা RAM ছাড়িয়ে গেলে চারটি পথ, সরল থেকে জটিল। (১) ছোট করুন: ঠিক dtype, দরকারি কলাম, Parquet. (২) টুকরো করুন: chunk-এ পড়ে প্রতিটি থেকে ছোট সারাংশ; model-এর ক্ষেত্রে out-of-core learning, অর্থাৎ ডিস্ক থেকে টুকরো টুকরো এনে শেখা। (৩) নমুনা নিন ধারণা যাচাইয়ের জন্য। (৪) টুল বা যন্ত্র বদলান: Polars বা DuckDB, সব core, আর সত্যিই দরকার হলে cluster. এই পর্ব এই ক্রমেই চলবে।

২ · প্রতি সারিতে কত byte — dtype দিয়ে memory-র হিসাব

একটি DataFrame-এর memory মোটামুটি সারির সংখ্যা × প্রতি সারির byte, আর তা আসে প্রতিটি কলামের dtype থেকে: int64 আর float64 ৮ byte, float32 ৪, uint8 আর bool ১. লেখার কলামে লেখাটি নিজে আর একটি ঠিকানা। pandas-এ এই আসল আকার দেখায় memory_usage(deep=True); deep ছাড়া লেখার কলামের ভেতরের অংশ গোনা হয় না। কমানোর তিনটি হাতিয়ার AI-এর ভিত্তি কোর্সের পর্ব ২৭-এ শেখানো: downcasting (সবচেয়ে ছোট যে dtype-এ সব মান ধরে), অল্প আলাদা মানের লেখায় category dtype, আর হ্যাঁ/না কলামে bool. নিচের ব্লক বাজারঘরের (কাল্পনিক দোকান) ১ লাখ অর্ডারে তিনটিই চালিয়ে প্রতি কলামের প্রতি সারির byte দেখায়।

import numpy as np
import pandas as pd

rng = np.random.default_rng(70)            # বাজারঘরের (কাল্পনিক) ১ লাখ অর্ডার
n = 100_000
DISTRICTS = ["Dhaka", "Chattogram", "Sylhet", "Rajshahi",
             "Khulna", "Barishal", "Rangpur", "Mymensingh"]
df = pd.DataFrame({
    "order_id": np.arange(n, dtype=np.int64),
    "district": rng.choice(DISTRICTS, n),
    "qty": rng.integers(1, 6, n, dtype=np.int64),
    "price": rng.lognormal(6, 0.8, n).round(2),
    "returned": rng.choice(["yes", "no"], n, p=[0.04, 0.96]),
})
small = df.assign(
    order_id=pd.to_numeric(df.order_id, downcast="unsigned"),
    district=df.district.astype("category"),
    qty=pd.to_numeric(df.qty, downcast="unsigned"),
    price=df.price.astype("float32"),
    returned=df.returned.eq("yes"))

per_row = lambda d: d.memory_usage(deep=True, index=False) / n
out = pd.DataFrame({"dtype_before": df.dtypes.astype(str),
                    "B/row": per_row(df).round(1),
                    "dtype_after": small.dtypes.astype(str),
                    "B/row_after": per_row(small).round(1)})
print(out)
print("total MiB:", round(per_row(df).sum() * n / 2**20, 2),
      "->", round(per_row(small).sum() * n / 2**20, 2))
err = (small.price - df.price).abs().max()
print("max price change after float32:", round(float(err), 5), "taka")
print("same revenue by district:", np.allclose(
    df.groupby("district").price.sum(),
    small.groupby("district", observed=True).price.sum(), rtol=1e-6))

         dtype_before  B/row dtype_after  B/row_after
order_id        int64    8.0      uint32          4.0
district          str   32.5    category          1.0
qty             int64    8.0       uint8          1.0
price         float64    8.0     float32          4.0
returned          str   27.0        bool          1.0
total MiB: 7.97 -> 1.05
max price change after float32: 0.00047 taka
same revenue by district: True

নিচের stepper প্রতিটি কলামের সিদ্ধান্ত আর ১ কোটি সারির হিসাব দেখায়; এমন হিসাব বিশ্ববিদ্যালয়, চাকরি বা ইন্টারভিউর পরীক্ষায় প্রায়ই আসে।

৩ · Chunk-এ হিসাব — কোন statistic জোড়া লাগে, কোনটি লাগে না

ছোট করার পরও না ধরলে ফাইল টুকরো করে পড়ুন: pd.read_csv(..., chunksize=70_000) একবারে ৭০ হাজার সারির একটি DataFrame দেয়। প্রতিটি chunk থেকে একটি ছোট সারাংশ রাখুন, আর শেষে সারাংশগুলো জুড়ুন। মূল প্রশ্ন: কোন সারাংশ জোড়া লাগানো যায়? যে সারাংশ দুটি অংশের সারাংশ থেকে পুরো অংশের সঠিক সারাংশ দেয়, তাকে বলে partial aggregate (জোড়া লাগানো যায় এমন সারাংশ)।

MapReduce (Google, ২০০৪) আর database-এর সমান্তরাল GROUP BY একই নিয়মে চলে: প্রতিটি অংশ আংশিক সারাংশ বানায়, তারপর সেগুলো মেলে। Variance-এর সারাংশ পরের অংশে; দুটির ব্লক অংশ ৪-এর শেষে।

৪ · Welford — এক পাসে mean আর variance

গড়ের জন্য এটি সহজ। Variance-এর জন্য পাঠ্যবইয়ের একটি সূত্র লোভনীয়: Σx আর Σx² জমান, শেষে variance = (Σx² − n·x̄²) ÷ (n − ১). সমস্যা হলো দুটি প্রায় সমান বিশাল সংখ্যার বিয়োগ। Floating-point সংখ্যা মোটামুটি ১৬টি অর্থবহ অঙ্ক রাখে; মানগুলো যদি ১০০ কোটির কাছে আর ছড়ানো মাত্র ১, তাহলে x² প্রায় ১০^১৮, আর আসল পার্থক্যটুকু শেষ অঙ্কগুলোর সঙ্গে হারিয়ে যায়। একে বলে catastrophic cancellation. ১৯৬২ সালে B. P. Welford Technometrics-এ একটি সমাধান দেন, যা আজ Welford-এর algorithm নামে পরিচিত। প্রতিটি নতুন মান x-এর জন্য (k = এ পর্যন্ত দেখা মানের সংখ্যা, mean = চলতি গড়, M2 = গড় থেকে বর্গ দূরত্বের চলতি যোগ):

k ← k + ১;  d = x − mean;  mean ← mean + d ÷ k;  M2 ← M2 + d · (x − নতুন mean)

শেষে sample variance = M2 ÷ (k − ১). এখানে কোনো বিশাল সংখ্যা জমে না; শুধু গড় থেকে দূরত্ব, যা ছোট। দুটি chunk-এর (n, mean, M2) জোড়া লাগাতে Chan, Golub আর LeVeque-এর (১৯৭৯) সূত্র আছে: δ = mean_B − mean_A, তারপর M2 = M2_A + M2_B + δ² · n_An_B ÷ (n_A + n_B). নিচের stepper চারটি সংখ্যায় Welford হাতে করে, যাতে খাতায়ও করতে পারেন।

1k = mean = m2 = 02for x in [4, 7, 13, 16]:3    k += 14    d = x - mean5    mean += d / k6    m2 += d * (x - mean)7var = m2 / (k - 1)

এবার দুটি ধারণা একসঙ্গে পরীক্ষা। ব্লকটি পাতার ভেতরেই একটি CSV লেখে (২ লাখ অর্ডার, তিনটি জেলা), তারপর তা ৭০ হাজার সারির chunk-এ পড়ে, আর পুরো ফাইল একবারে পড়ার উত্তরের সঙ্গে মেলায়। শেষে ১০০ কোটির কাছের ১ লাখ মানে Welford, NumPy আর পাঠ্যবইয়ের সূত্রের তুলনা।

import numpy as np
import pandas as pd

rng = np.random.default_rng(70)               # বাজারঘরের (কাল্পনিক) ২ লাখ অর্ডার
n = 200_000
pd.DataFrame({"district": rng.choice(["Dhaka", "Khulna", "Sylhet"], n, p=[.6, .25, .15]),
              "price": rng.lognormal(6, 0.8, n).round(2)}).to_csv("orders.csv", index=False)

parts, means = [], []
for chunk in pd.read_csv("orders.csv", chunksize=70_000):   # একবারে ৭০ হাজার সারি
    g = chunk.groupby("district").price
    parts.append(g.agg(["count", "sum"]))     # জোড়া লাগানো যায়: count, sum
    means.append(g.mean())                    # জোড়া লাগানো যায় না: mean
agg = pd.concat(parts).groupby(level=0).sum()
agg["mean"] = agg["sum"] / agg["count"]
full = pd.read_csv("orders.csv").groupby("district").price.agg(["count", "sum", "mean"])
print(len(parts), "chunks; same as full read:", np.allclose(agg, full))
print(agg.round(2))
print("mean of chunk means:", pd.concat(means).groupby(level=0).mean().round(2).to_dict())

def welford(xs):                              # এক পাসে mean আর variance
    k = mean = m2 = 0.0
    for x in xs:
        k += 1
        d = x - mean                          # পুরোনো mean থেকে দূরত্ব
        mean += d / k
        m2 += d * (x - mean)                  # পুরোনো আর নতুন দূরত্বের গুণফল
    return mean, m2 / (k - 1)

x = 1e9 + rng.normal(0, 1, 100_000)           # বড় মান, ছোট ছড়ানো
naive = (np.sum(x**2) - len(x) * x.mean()**2) / (len(x) - 1)
m, v = welford(x.tolist())
print(f"welford: mean={m:.4f}  var={v:.6f}")
print(f"numpy  : mean={x.mean():.4f}  var={x.var(ddof=1):.6f}")
print("naive sum-of-squares variance:", naive)

3 chunks; same as full read: True
           count          sum    mean
district                             
Dhaka     120309  66918292.63  556.22
Khulna     49740  27641208.70  555.71
Sylhet     29951  16629507.18  555.22
mean of chunk means: {'Dhaka': 556.13, 'Khulna': 555.51, 'Sylhet': 555.41}
welford: mean=999999999.9981  var=0.991460
numpy  : mean=999999999.9981  var=0.991460
naive sum-of-squares variance: 0.0

৫ · Sampling — প্রথমে ছোট নমুনায় ধারণা যাচাই

পুরো ডেটায় একটি ধারণা পরীক্ষায় এক ঘণ্টা লাগলে দিনে আটটির বেশি ধারণা যাচাই হয় না। তাই প্রথমে একটি ছোট, ভালোভাবে বাছা নমুনায় feature, model আর pipeline ঠিক করুন; পুরো ডেটা আসে শেষে। কয়েক লাখ সারিতে learning curve (AI-এর ভিত্তি কোর্সের পর্ব ৬৬) যদি সমতল হয়ে যায়, তাহলে পুরো ১০ কোটি সারি নিয়ে কষ্ট করার দরকারই নাও হতে পারে।

• Simple random sampling: প্রতিটি সারির বাছাইয়ের সম্ভাবনা সমান। DuckDB-র USING SAMPLE 1% ফাইল থেকেই নমুনা নিতে পারে; DataFrame-এ pandas আর Polars-এর sample. Seed লিখে রাখুন (পর্ব ০৪)।
• Stratified sampling: জেলা বা শ্রেণি ধরে আলাদা নমুনা। ফেরতের হার ৪% হলে সরল ১% নমুনায় ছোট জেলার ফেরত-অর্ডার হাতে গোনা কয়েকটি আসতে পারে।
• বিরল শ্রেণির downsampling: সব fraud রেখে সাধারণ অর্ডারের মাত্র ৫% রাখা। কিন্তু তখন model-এর সম্ভাবনা আর আসল হার নয়; সংশোধন আর threshold পর্ব ৫০-এর বিষয়।
• Reservoir sampling: ডেটা স্রোতের মতো আসছে, মোট কত সারি জানা নেই, তবু ঠিক k-টির একটি সমান-সম্ভাবনার নমুনা চাই। প্রথম k-টি রাখুন; তারপর i-তম সারিকে (০ থেকে গুনে) k ÷ (i + ১) সম্ভাবনায় নমুনার একটি এলোমেলো জায়গায় বসান। এক পাসে, k-টির বেশি memory ছাড়া। Jeffrey Vitter ১৯৮৫ সালে এর বিশ্লেষণ আর দ্রুত রূপ প্রকাশ করেন; সরল রূপটি (Algorithm R) Donald Knuth-এর বইয়ে Alan Waterman-এর নামে উল্লেখিত। অনুশীলন ৪-এ নিজে লিখবেন।

৬ · Incremental learning — partial_fit দিয়ে টুকরো টুকরো শেখা

সারাংশ জোড়া লাগানো যায়, কিন্তু model? অনেক model-এর জন্য উত্তর হ্যাঁ, যদি শেখার নিয়মটি নিজেই একবারে অল্প ডেটা দেখে। Stochastic gradient descent ঠিক তাই করে: প্রতিটি mini-batch-এর gradient দিয়ে একটি ছোট পা (পর্ব ১৩; বড় ডেটায় kernel-এর বদলে linear model আর SGD কেন, তা পর্ব ৩১-এ)। scikit-learn-এ এর দরজা partial_fit: fit পুরোনো সব ভুলে নতুন করে শেখে, আর partial_fit যা শিখেছে তার ওপর একটি chunk-এর এক পাস যোগ করে। এভাবে শেখানোকে বলে incremental learning.

partial_fit আছে, এমন কয়েকটি: SGDClassifier, SGDRegressor, Perceptron, MultinomialNB আর অন্য Naive Bayes, MiniBatchKMeans, IncrementalPCA, MLPClassifier, আর preprocessing-এ StandardScaler, MinMaxScaler. Random forest বা HistGradientBoosting-এর নেই (XGBoost-এর external memory-র মতো আলাদা ব্যবস্থা আছে, এই পর্বের বাইরে)। MiniBatchKMeans এই পাতায় চালানো হয়নি, কারণ ব্রাউজারের Python-এ KMeans পরিবার একটি threadpoolctl সতর্কবার্তা ছাপে।

লেখার feature-এ আরেক সমস্যা: TfidfVectorizer আগে পুরো ডেটা দেখে শব্দভান্ডার বানায়। সমাধান feature hashing (AI-এর ভিত্তি কোর্সের পর্ব ৬৩): প্রতিটি শব্দকে একটি hash function দিয়ে নির্দিষ্ট সংখ্যক কলামের একটিতে পাঠানো। HashingVectorizer কিছুই শেখে না, তাই নতুন শব্দও প্রতিবার একই কলামে যায়। দাম দুটি: কখনো দুটি শব্দ এক কলামে পড়ে (hash collision), আর কলাম থেকে শব্দে ফেরা যায় না। বাংলা শব্দের কার-চিহ্ন regex-এর \w-এ ঠিকমতো ধরা পড়ে না, তাই এখানে ফাঁকা জায়গা ধরে ভাগ করা হয়েছে (token_pattern=r"\S+"; বাংলা tokenization AI-এর ভিত্তি কোর্সের পর্ব ৬৭-এ)।

নিচের ব্লক বাজারঘরের বানানো পণ্য-শিরোনাম (যেমন “অফার বড় কাগুগু”; শেষ শব্দটি বানানো brand) আর দাম থেকে পণ্যের শ্রেণি অনুমান করে। প্রতি শ্রেণিতে ১৫০টি বানানো brand, তাই শুরুর chunk-গুলো সব brand দেখেনি। ১০ হাজার শিরোনাম আসে ১,০০০-এর ১০টি chunk-এ; প্রতিটির পর একটি স্থির ৪,০০০ শিরোনামের test set-এ accuracy মাপা হয়। দামের StandardScaler-ও partial_fit দিয়ে ধাপে ধাপে শেখে। তুলনা: সব ১০ হাজার সারি একসঙ্গে RAM-এ রেখে সাধারণ fit, যা ডেটার ওপর অনেকবার ঘোরে (অনেক epoch).

import numpy as np
import matplotlib.pyplot as plt
from scipy import sparse
from sklearn.feature_extraction.text import HashingVectorizer
from sklearn.linear_model import SGDClassifier
from sklearn.preprocessing import StandardScaler

rng = np.random.default_rng(70)     # বাজারঘরের (কাল্পনিক) পণ্যের শিরোনাম থেকে category
WORDS = [["চাল", "ডাল", "আটা", "চিনি"], ["শাড়ি", "পাঞ্জাবি", "কুর্তি", "জুতা"],
         ["মোবাইল", "চার্জার", "হেডফোন", "ল্যাপটপ"], ["উপন্যাস", "অভিধান", "কবিতা", "গাইড"]]
SYL = ["কা", "রি", "মো", "তু", "লে", "সা", "নি", "বো", "দা", "পি", "গু", "হে"]
BRANDS = [["".join(rng.choice(SYL, 3)) for _ in range(150)] for _ in WORDS]  # বানানো brand
EXTRA = ["নতুন", "অফার", "সেরা", "প্রিমিয়াম", "ছাড়", "আসল", "বড়", "ছোট"]
def make(n):
    y = rng.integers(0, 4, n)
    t = [" ".join([*rng.choice(EXTRA, 2), rng.choice(BRANDS[c])] +
                  ([rng.choice(WORDS[c])] if rng.random() < 0.3 else [])) for c in y]
    return t, rng.normal(np.array([6.0, 7.0, 7.5, 5.5])[y], 0.9)[:, None], y
(T, P, Y), (Tt, Pt, Yt) = make(10_000), make(4_000)
hv = HashingVectorizer(n_features=2**12, token_pattern=r"\S+", alternate_sign=False)
X = lambda t, p, sc: sparse.hstack([hv.transform(t), sc.transform(p)]).tocsr()

sc, clf, acc = StandardScaler(), SGDClassifier(loss="log_loss", random_state=0), []
for s in range(0, 10_000, 1_000):                 # ১০টি chunk, প্রতিটি একবার দেখা
    sc.partial_fit(P[s:s + 1000])                 # scaler-এর গড়ও ধাপে ধাপে
    clf.partial_fit(X(T[s:s + 1000], P[s:s + 1000], sc), Y[s:s + 1000], classes=range(4))
    acc.append(clf.score(X(Tt, Pt, sc), Yt))
sc2 = StandardScaler().fit(P)                     # তুলনা: সব সারি একসঙ্গে, অনেক epoch
full = SGDClassifier(loss="log_loss", random_state=0).fit(X(T, P, sc2), Y)
print("example title:", T[0], "->", ["grocery", "fashion", "electronics", "books"][Y[0]])
print("after each chunk:", np.round(acc, 3))
fa = full.score(X(Tt, Pt, sc2), Yt); print("full batch, all rows in memory:", round(fa, 3))
plt.plot(range(1000, 10_001, 1000), acc, "o-", label="partial_fit, one pass")
plt.axhline(fa, ls="--", color="gray", label="full-batch fit")
plt.xlabel("rows seen"); plt.ylabel("test accuracy"); plt.legend(); plt.show()

example title: অফার বড় কাগুগু -> fashion
after each chunk: [0.691 0.707 0.782 0.757 0.824 0.878 0.877 0.904 0.906 0.892]
full batch, all rows in memory: 0.938

৭ · Parquet — কলামে রাখা, row group আর min-max

CSV প্রতিটি সারি একটি লাইনে লেখে। দুটি কলাম লাগলেও প্রতিটি লাইন পুরো পড়ে ভাঙতে হয়, আর প্রতিটি সংখ্যা লেখা থেকে আবার সংখ্যায় রূপান্তর করতে হয়। Parquet উল্টো: columnar storage, অর্থাৎ একই কলামের মান পাশাপাশি, binary-তে, dtype সহ, চাপা (compressed) অবস্থায়। ফাইলটি অনেকগুলো row group-এ ভাগ (প্রতিটি কয়েক হাজার থেকে কয়েক লাখ সারি), আর ফাইলের শেষে একটি footer-এ প্রতিটি row group-এর প্রতিটি কলামের column statistics লেখা: min, max, null-এর সংখ্যা।

পাতার Python-এ Parquet লেখা হয় pyarrow দিয়ে (pyarrow.parquet.write_table; এখানে df.to_parquet চলে না)। নিচের ব্লক ৩ লাখ অর্ডার দিনের ক্রমে লেখে, প্রতি ৫০ হাজার সারিতে একটি row group. তারপর দুই ফাইলের আকার আর দুটি কলাম পড়ার সময়, footer-এর min-max, আর একটি filter. শেষে একই প্রশ্ন (electronics-এর জেলাভিত্তিক আয়) তিনটি টুলে: pandas, Polars-এর lazy পথ আর DuckDB-র SQL. সময় একবার গরম করে নেওয়ার পর দ্বিতীয় run-এর; প্রতিবার চালালে একটু বদলাবে।

import os, time
import duckdb, numpy as np, pandas as pd, polars as pl
import pyarrow as pa, pyarrow.parquet as pq

rng = np.random.default_rng(70); n = 300_000   # দিনের ক্রমে আসা অর্ডার (কাল্পনিক)
df = pd.DataFrame({"day": np.sort(rng.integers(0, 365, n)),
    "district": rng.choice(["Dhaka", "Khulna", "Sylhet", "Rangpur"], n),
    "category": rng.choice(["grocery", "fashion", "electronics"], n),
    "price": rng.lognormal(6, 0.8, n).round(2), "qty": rng.integers(1, 6, n)})
df.to_csv("orders.csv", index=False)
pq.write_table(pa.Table.from_pandas(df, preserve_index=False), "orders.parquet",
               row_group_size=50_000)              # প্রতি ৫০ হাজার সারিতে একটি row group
def ms(f):                                         # দ্বিতীয়বারের সময় (প্রথমবার গরম করা)
    f(); t = time.perf_counter(); r = f(); return r, 1000 * (time.perf_counter() - t)
print({k: round(os.path.getsize(k) / 2**20, 2) for k in ["orders.csv", "orders.parquet"]}, "MiB")
_, t1 = ms(lambda: pd.read_csv("orders.csv", usecols=["district", "price"]))
_, t2 = ms(lambda: pq.read_table("orders.parquet", columns=["district", "price"]))
print(f"read 2 of 5 columns: CSV {t1:.0f} ms, Parquet {t2:.0f} ms")
meta = pq.ParquetFile("orders.parquet").metadata
print("day min-max per row group:", [(meta.row_group(i).column(0).statistics.min,
      meta.row_group(i).column(0).statistics.max) for i in range(meta.num_row_groups)])
print("rows with day >= 330:", pq.read_table("orders.parquet", filters=[("day", ">=", 330)]).num_rows)

cols = lambda: pq.read_table("orders.parquet", columns=["district", "category", "price", "qty"])
def with_pandas():                                 # তিন টুলে একই প্রশ্ন, একই ফাইল থেকে
    d = cols().to_pandas()
    return d[d.category == "electronics"].assign(rev=d.price * d.qty).groupby("district").rev.sum()
a, ta = ms(with_pandas)
b, tb = ms(lambda: pl.from_arrow(cols()).lazy().filter(pl.col("category") == "electronics")
           .group_by("district").agg((pl.col("price") * pl.col("qty")).sum().alias("rev"))
           .sort("district").collect())
c, tc = ms(lambda: duckdb.sql("SELECT district, sum(price * qty) AS rev FROM 'orders.parquet' "
           "WHERE category = 'electronics' GROUP BY district ORDER BY district").df())
print(f"pandas {ta:.0f} ms, Polars lazy {tb:.0f} ms, DuckDB {tc:.0f} ms")
print("same answer:", np.allclose(a, b["rev"].to_numpy()) and np.allclose(a, c.rev), a.round().to_dict())

{'orders.csv': 8.3, 'orders.parquet': 1.94} MiB
read 2 of 5 columns: CSV 96 ms, Parquet 33 ms
day min-max per row group: [(0, 60), (60, 121), (121, 181), (181, 242), (242, 303), (303, 364)]
rows with day >= 330: 28694
pandas 196 ms, Polars lazy 74 ms, DuckDB 102 ms
same answer: True {'Dhaka': 41689747.0, 'Khulna': 41961071.0, 'Rangpur': 41387004.0, 'Sylhet': 41695871.0}

ভালো Parquet-এর তিন অভ্যাস: (ক) যে কলামে বেশি filter হয় (তারিখ), সেই ক্রমে লিখুন। (খ) row group খুব ছোট নয় (footer বড়, পড়া ধীর), খুব বড়ও নয় (একটি খুলতেই অনেক memory); pyarrow-এর ডিফল্ট সীমা প্রায় ১০ লাখ সারি। (গ) খুব বড় ডেটা মাস ধরে আলাদা ফোল্ডারে (month=2026-10/), যাতে ফোল্ডারের নাম দেখেই পুরো মাস বাদ যায়; একে partitioning বলে (System Design কোর্সের partitioning পর্ব)।

৮ · এক কোটি সারি — pandas, Polars আর DuckDB-র আসল মাপ

পাতার ৩ লাখ সারি টুলের পার্থক্য দেখাতে খুব ছোট। তাই ডেস্কটপে একটি আলাদা uv project-এ (AI-এর ভিত্তি কোর্সের পর্ব ১২) seed দিয়ে বাজারঘরের ১ কোটি অর্ডার বানানো হয়েছে: আটটি জেলা, ছয়টি শ্রেণি, দাম, পরিমাণ, অনলাইনে পেমেন্ট আর ফেরত। লেখার সময়ে Python 3.13.5, pandas 3.1.1, Polars 2.1.0, DuckDB 2.0.0, pyarrow 26.0.1, psutil 7.2.2. CSV ৫১,৬০,১৪,০৫৪ byte (৪৯২ MiB), Parquet ১১,১২,৪৭,০৪৯ byte (১০৬ MiB, ১০টি row group). প্রশ্ন: ফেরত হয়নি এমন electronics অর্ডারে প্রতি জেলায় অর্ডার, মোট আয় আর গড় দাম। Polars-এর lazy query আর তার explain() AI-এর ভিত্তি কোর্সের পর্ব ২৭-এ ধাপে ধাপে আছে; এখানে নতুন মাপ memory.

METHODS = {
    "pandas_csv": lambda: query_pandas(pd.read_csv("orders.csv")),
    "pandas_parquet": lambda: query_pandas(pd.read_parquet("orders.parquet", columns=COLS)),
    "polars_lazy": lambda: lazy.collect().to_pandas(),   # pl.scan_parquet(...) থেকে
    "duckdb_sql": lambda: duckdb.sql(SQL).df(),           # FROM 'orders.parquet'
}
proc = psutil.Process()
base = proc.memory_info().rss                 # import-এর পরের memory
for _ in range(3):                            # তিনবার, প্রতিটির সময়
    t0 = time.perf_counter(); METHODS[sys.argv[1]](); ts.append(time.perf_counter() - t0)
peak = proc.memory_info().peak_wset - base    # Windows-এ সর্বোচ্চ working set

~/bazarghar-scale$ for m in pandas_csv pandas_parquet polars_lazy duckdb_sql; do uv run python bench.py $m; done
pandas_csv      first  0.74 s  median  0.74 s  extra peak memory   1645 MiB
pandas_parquet  first  0.54 s  median  0.46 s  extra peak memory   1119 MiB
polars_lazy     first  0.04 s  median  0.02 s  extra peak memory    252 MiB
duckdb_sql      first  0.08 s  median  0.04 s  extra peak memory    108 MiB
~/bazarghar-scale$ uv run python bench.py check
python 3.13.5  pandas 3.1.1  polars 2.1.0  duckdb 2.0.0
  district  orders     revenue  avg_price
     Dhaka  561440 933848042.2      554.5
Chattogram  287418 478261585.0      554.7
    Khulna  160086 267040535.5      557.9
…
  Barishal   95755 159620970.4      558.2
same answer in all four: True

RAM-এর চেয়েও বড় ডেটায় দুটিরই উত্তর আছে। DuckDB-র অনেক operator (group by, join, sort) memory-র সীমা ছুঁলে অস্থায়ী ফল ডিস্কে লেখে (spill to disk) আর কাজ চালিয়ে যায়; সীমা SET memory_limit = '4GB'. Polars-এর streaming engine lazy query টুকরো টুকরো চালায়। সব operator সমানভাবে পারে না, তাই নিজের version-এর docs দেখুন। ML-এর সাধারণ ধারা: কাঁচা টেবিল SQL বা Polars দিয়ে ক্রেতা বা দিন প্রতি ছোট feature টেবিলে নামান, Parquet-এ রাখুন, তারপর pandas আর scikit-learn. SQL-এর window function আর cohort পর্ব ৬১-এ।

৯ · সব core, তারপর cluster — আর টুল বাছাইয়ের সিদ্ধান্ত

এক যন্ত্রের ভেতরে প্রথম ধাপ সব CPU core কাজে লাগানো। scikit-learn-এর অনেক estimator আর cross_val_score, GridSearchCV-এর মতো টুল n_jobs নেয়, আর ভেতরে joblib কাজ ভাগ করে: random forest-এর আলাদা গাছ, বা cross-validation-এর আলাদা fold আলাদা core-এ (data parallelism). পাতার Python-এ একটিই thread, তাই এটি ডেস্কটপের সত্যিকারের run (scikit-learn 1.9.1, ১ লাখ সারি, ২০টি feature, ১০০টি গাছ):

~/bazarghar-scale$ uv run python njobs.py
sklearn 1.9.1  n_jobs=1:  35.2 s, train acc 1.000
sklearn 1.9.1  n_jobs=4:   9.3 s, train acc 1.000

চারটি core-এ প্রায় ৩.৮ গুণ দ্রুত, আর একই random_state-এ গাছগুলো একই (train accuracy শুধু মিল দেখাতে, মূল্যায়ন নয়)। তবে n_jobs=-1 চোখ বন্ধ করে নয়: প্রতিটি worker process-এর নিজের memory লাগে, আর বড় ডেটা প্রতিটিতে copy হলে RAM শেষ। GridSearchCV(n_jobs=8)-এর ভেতরে model-ও n_jobs=8 হলে ৬৪টি কাজ আটটি core-এর জন্য লড়ে। একটি স্তরেই সমান্তরাল করুন।

এক যন্ত্র ছাড়িয়ে গেলে Dask আর Apache Spark-এর মতো টুল কাজ অনেক যন্ত্রে ভাগ করে (ডেটা পাইপলাইনের নকশা System Design কোর্সের batch আর stream processing পর্ব-এ)। Dask Python-এর ভেতরে pandas-এর মতো API দেয়, আর একই কোড ল্যাপটপে বা cluster-এ চালাতে পারে; Spark JVM-ভিত্তিক, বড় প্রতিষ্ঠানের ডেটা পাইপলাইনে বহুল প্রচলিত, Python থেকে PySpark দিয়ে ব্যবহার হয়। দুটিই শক্তিশালী, কিন্তু নতুন জটিলতা আনে: নেটওয়ার্কে ডেটা সরানোর খরচ (shuffle), যন্ত্র বিকল হওয়া, খরচ আর debug-এর কষ্ট; আর সব ML algorithm ভাগ করা সহজ নয়। অনেক দলের (সবার নয়) পরামর্শ: cluster-এ ডেটা ছোট feature টেবিলে নামান, শেখান এক যন্ত্রে। Cluster বানানো আর চালানো একটি পরের MLOps আর ডেটা ইঞ্জিনিয়ারিং কোর্সের বিষয়।

নিচের stepper তিন আকারের ডেটায় সিদ্ধান্তটি দেখায়। ধরুন যন্ত্রটি ১৬ GB RAM-এর একটি ল্যাপটপ।

১০ · AI-এর যুগে — বড় ডেটার ML কোড যাচাই

AI assistant বড় ডেটার কোড দ্রুত দেয়, যা প্রায়ই ছোট ডেটায় নিখুঁত চলে; ভুল ধরা পড়ে বড় ফাইলে, বা থেকে যায় চুপচাপ ভুল সংখ্যায়।

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

• ছোট একটি অংশে chunk-এর উত্তর পুরো-পড়ার উত্তরের সঙ্গে হুবহু মিলিয়েছি
• প্রতিটি method (partial_fit, Polars-এর নাম) নিজের version-এর docs-এ আছে কি না দেখেছি
• প্রতি সারির byte হিসাব করে memory অনুমান করেছি আর সর্বোচ্চ memory মেপেছি
• Test set শুরুতে আলাদা আর স্থির; অন্তত একটি সরল baseline আর কয়েকটি seed-এর সঙ্গে তুলনা করেছি
• Variance বা SD-র কোড বড় মানে (১০^৯ যোগ করে) পরীক্ষা করেছি
• আসল ক্রেতার ডেটা AI-তে পেস্ট করিনি; schema আর বানানো সারি দিয়েছি