Batch আর stream processing — MapReduce, window আর watermark

১ · Batch আর stream — সীমিত ডেটা বনাম অসীম ধারা

পাড়ার একটি মুদি দোকানের কথা ভাবুন। দোকানি দুইভাবে হিসাব রাখেন। প্রতি মাসের শেষে খাতা খুলে পুরো মাসের বিক্রি যোগ করেন — কোন পণ্য বেশি চলল, কার কাছে কত বাকি। এই হিসাব নিখুঁত, কারণ মাস শেষ, আর কোনো বিক্রি যোগ হবে না। কিন্তু উত্তর আসে মাসে একবার। আবার দিনের মধ্যেও তিনি ক্যাশবাক্সে চোখ রাখেন: “গত এক ঘণ্টায় চিনি অনেক বিক্রি হলো, আড়তে এখনই ফোন দিই।” এই হিসাব তাৎক্ষণিক, কিন্তু কখনো পুরো নিশ্চিত নয় — কোনো খদ্দের হয়তো বাকিতে নিয়েছেন, টাকা দেবেন পরে।

প্রথমটি batch processing: একটি সীমিত, শেষ হয়ে যাওয়া ডেটার স্তূপ (bounded data) নিয়ে একবারে হিসাব। দ্বিতীয়টি stream processing: যে ডেটা কখনো শেষ হয় না (unbounded data) — অ্যাপের click, পেমেন্ট, GPS-এর অবস্থান — তার ওপর চলমান হিসাব। দুটোরই কাজ কাঁচা ঘটনা থেকে কাজের তথ্য বানানো; পার্থক্য হলো কখন উত্তর চাই আর উত্তর কতটা সম্পূর্ণ হতে হবে।

একটি গুরুত্বপূর্ণ উপলব্ধি: batch আসলে stream-এর একটি বিশেষ রূপ। গতকালের log হলো একটি stream, যার শেষ আছে। তাই আধুনিক অনেক engine একই API দিয়ে দুটোই চালায়। উল্টোটা সত্য নয় — অসীম ডেটাকে batch-এ চালাতে হলে কৃত্রিমভাবে টুকরো (যেমন প্রতি ঘণ্টা) করতে হয়, আর টুকরোর সীমায় পড়া ঘটনা (রাত ১১:৫৯-এ শুরু হওয়া একটি session) ভেঙে যায়। এই কথাটি অংশ ৫-এ আবার আসবে।

নিচের ছবিতে একটি সাধারণ বড় সিস্টেমের দুই পথ পাশাপাশি দেখুন। ঘটনাগুলো প্রথমে একটি log-এ জমে (পর্ব ৩৩); সেখান থেকে stream processor সেকেন্ডে উত্তর দেয়, আর একই ডেটা data lake-এ জমে রাতের batch কাজে যায়।

বাস্তবে কোন পথ লাগবে, সেটি আসে requirement থেকে (পর্ব ২)। “মাসিক বিক্রির report” — batch যথেষ্ট, সস্তাও। “সন্দেহজনক লেনদেন ২ সেকেন্ডের মধ্যে আটকাতে হবে” — stream ছাড়া উপায় নেই।

২ · MapReduce — map, shuffle আর reduce

ডাকঘরে চিঠি বাছাইয়ের দৃশ্য মনে করুন। সারা দেশ থেকে লাখো চিঠি আসে। এক দল কর্মী প্রতিটি চিঠি হাতে নিয়ে শুধু জেলার নামটি পড়েন আর একটি চিহ্ন দেন — এটি map. তারপর চিঠিগুলো জেলা অনুযায়ী বস্তায় ভরে আলাদা আলাদা টেবিলে পাঠানো হয়: “রাজশাহীর সব চিঠি তিন নম্বর টেবিলে” — এটি shuffle. শেষে প্রতিটি টেবিলের কর্মী নিজের জেলার চিঠিগুলো গুনে বা সাজিয়ে পরের ধাপে পাঠান — এটি reduce. কেউ পুরো স্তূপ একা দেখেন না, তবু শেষে সব চিঠি ঠিক জায়গায়।

২০০৪ সালে Google-এর Jeffrey Dean ও Sanjay Ghemawat OSDI সম্মেলনে “MapReduce: Simplified Data Processing on Large Clusters” প্রবন্ধে ঠিক এই ভাবনাটিকে একটি প্রোগ্রামিং মডেল হিসেবে প্রকাশ করেন। প্রোগ্রামার লেখেন শুধু দুটি ফাংশন; ভাগ করা, হাজার মেশিনে চালানো, মেশিন বিকল হলে আবার চালানো — সব করে framework. প্রবন্ধে বর্ণিত MapReduce মডেলটি এরকম:

Map ধাপে (map phase) প্রতিটি map task ইনপুটের একটি টুকরো (split) পড়ে, আর প্রতিটি রেকর্ড থেকে শূন্য বা বেশি (key, value) জোড়া বানায়। Shuffle ধাপে প্রতিটি জোড়া hash দেখে R-টি reducer-এর একটিতে যায়, আর প্রতিটি reducer নিজের জোড়াগুলো key অনুযায়ী sort করে দল বাঁধে। Reduce ধাপে (reduce phase) প্রতিটি key-এর সব value একসাথে পেয়ে ফল লেখা হয়। সবচেয়ে চেনা উদাহরণ — কোন শব্দ কতবার এসেছে (word count). নিচের stepper-এ দুটি map task আর দুটি reducer দিয়ে পুরো পথ দেখুন:

split 0 → map task 0
  "ঢাকা rain ঢাকা traffic"
  "rain ঢাকা rain"
split 1 → map task 1
  "traffic jam ঢাকা"
  "jam jam rain"

task 0: (ঢাকা,1) (rain,1) (ঢাকা,1) (traffic,1)
        (rain,1) (ঢাকা,1) (rain,1)
task 1: (traffic,1) (jam,1) (ঢাকা,1)
        (jam,1) (jam,1) (rain,1)

task 0: (ঢাকা,3) (rain,3) (traffic,1)
task 1: (traffic,1) (jam,3) (ঢাকা,1) (rain,1)

reducer 0 ← traffic, ঢাকা   (দুই task থেকেই)
reducer 1 ← jam, rain
একই শব্দ সবসময় একই reducer-এ

reducer 0: traffic [1,1]   ঢাকা [3,1]
reducer 1: jam [3]         rain [3,1]

part-00000: traffic=2  ঢাকা=4
part-00001: jam=3      rain=4
মোট শব্দ 13 = 2 + 4 + 3 + 4

এবার একই কাজটি নিজে চালিয়ে দেখুন। কোডে একটি প্রক্রিয়ার ভেতরে দুটি map task আর দুটি reducer সাজানো হয়েছে; partition ঠিক হয় FNV-1a hash দিয়ে (FNV-1a, পর্ব ১৬-এ দেখেছেন)।

// MapReduce word count on 2 machines (map tasks) and 2 reducers, in one process.
const splits = [                                  // each split = one map task's input
  ["ঢাকা rain ঢাকা traffic", "rain ঢাকা rain"],
  ["traffic jam ঢাকা", "jam jam rain"] ];
const R = 2;                                      // number of reduce tasks
function fnv1a(s) {                               // FNV-1a 32-bit over UTF-16 units
  let h = 0x811c9dc5;
  for (const ch of s) { h ^= ch.charCodeAt(0); h = Math.imul(h, 0x01000193) >>> 0; }
  return h;
}
const part = (w) => fnv1a(w) % R;                 // partition function: hash(key) mod R
const mapFn = (line) => line.split(" ").map((w) => [w, 1]);   // emit (word, 1)
const combine = (pairs) => {                      // local reduce on the map machine
  const m = new Map(); for (const [w, c] of pairs) m.set(w, (m.get(w) || 0) + c);
  return [...m];
};
let sentPlain = 0, sentComb = 0;
const buckets = Array.from({ length: R }, () => new Map());
splits.forEach((lines, i) => {
  const out = lines.flatMap(mapFn), comb = combine(out);
  sentPlain += out.length; sentComb += comb.length;
  console.log(`map task ${i}: ` + out.map(([w, c]) => `(${w},${c})`).join(" "));
  console.log(`  after combiner: ` + comb.map(([w, c]) => `(${w},${c})`).join(" "));
  for (const [w, c] of comb) {                    // shuffle: send to reducer part(w)
    const b = buckets[part(w)]; if (!b.has(w)) b.set(w, []); b.get(w).push(c);
  }
});
buckets.forEach((b, r) => {
  const keys = [...b.keys()].sort();              // each reducer sees its keys sorted
  console.log(`\nreducer ${r} gets: ` + keys.map((k) => `${k} [${b.get(k)}]`).join("  "));
  console.log(`  output: ` + keys.map((k) => `${k}=${b.get(k).reduce((a, x) => a + x, 0)}`).join("  "));
});
console.log(`\npairs shuffled: ${sentPlain} without combiner, ${sentComb} with combiner`);

map task 0: (ঢাকা,1) (rain,1) (ঢাকা,1) (traffic,1) (rain,1) (ঢাকা,1) (rain,1)
  after combiner: (ঢাকা,3) (rain,3) (traffic,1)
map task 1: (traffic,1) (jam,1) (ঢাকা,1) (jam,1) (jam,1) (rain,1)
  after combiner: (traffic,1) (jam,3) (ঢাকা,1) (rain,1)

reducer 0 gets: traffic [1,1]  ঢাকা [3,1]
  output: traffic=2  ঢাকা=4

reducer 1 gets: jam [3]  rain [3,1]
  output: jam=3  rain=4

pairs shuffled: 13 without combiner, 7 with combiner

৩ · হাজার মেশিনে MapReduce — data locality, straggler আর ব্যর্থতা

মডেলটি সরল, কিন্তু হাজার মেশিনে চালাতে গেলে তিনটি বাস্তব সমস্যা আসে: ডেটা নড়াচড়ার খরচ, ধীর মেশিন, আর বিকল মেশিন। মূল প্রবন্ধের অনেকটাই এই তিনটির সমাধান।

Data locality — কাজকে ডেটার কাছে পাঠানো

ইনপুট থাকে distributed file system-এ — Google-এ GFS, open-source জগতে HDFS (দুটোই পর্ব ২৬-এ দেখেছেন)। প্রতিটি block-এর সাধারণত ৩টি কপি তিনটি মেশিনে। Scheduler তাই map task সেই মেশিনেই চালাতে চায় যেখানে block-এর একটি কপি আছে; না পারলে একই rack-এর মেশিনে। কয়েক MB কোড পাঠানো, কয়েকশো GB ডেটা টানার চেয়ে অনেক সস্তা। এর নাম data locality.

কিন্তু shuffle-এ locality নেই: প্রতিটি map task-এর output প্রতিটি reducer-এর কাছে যেতে পারে, অর্থাৎ M × R জোড়া সংযোগ। তাই বড় কাজে সবচেয়ে ব্যয়বহুল ধাপ সাধারণত shuffle — disk-এ লেখা, নেটওয়ার্কে পাঠানো, আবার sort করা। Combiner, compression আর ভালো partition — সবই shuffle ছোট করার চেষ্টা।

Straggler আর speculative execution

রেলস্টেশনের দশটি টিকিট কাউন্টারের নয়টিতে লাইন শেষ, একটিতে কম্পিউটার ধীর — পুরো যাত্রীদল সেই একটির জন্য দাঁড়িয়ে। MapReduce-এ কাজ শেষ হয় তখনই, যখন সবচেয়ে ধীর task শেষ হয়। খারাপ disk, ব্যস্ত CPU বা ভুল configuration-এর একটি মেশিন — একটি straggler — পুরো কাজ আটকে রাখে। সমাধান: কাজ শেষের দিকে এলে বাকি চলমান task-গুলোর একটি করে বাড়তি কপি (backup task) অন্য মেশিনে চালানো; যেটি আগে শেষ হয়, তার ফল নেওয়া, অন্যটি বাতিল। Hadoop-এ এর নাম speculative execution. প্রবন্ধের একটি পরীক্ষায় এই ব্যবস্থা বন্ধ করলে sort কাজ প্রায় ৪৪% বেশি সময় নেয় — কয়েক শতাংশ বাড়তি কাজের দামে বড় সাশ্রয়। এটি পর্ব ৪-এর hedged request-এর একই চিন্তা।

Speculative execution সব straggler সারায় না। কোনো key-তে অস্বাভাবিক বেশি ডেটা থাকলে — যেমন word count-এ “the” বা “এবং”, অথবা একটি ভাইরাল পোস্টের id — সেই reducer-টি ধীর হয় মেশিনের দোষে নয়, ডেটার ভারে। বাড়তি কপিও একই ভার টানবে। এটি data skew বা hot key-এর সমস্যা (পর্ব ২৩)। প্রতিকার: combiner, key-কে লবণ মিশিয়ে কয়েক টুকরো করা (key salting) আর শেষে টুকরোগুলো মেলানো।

ব্যর্থতা — আবার চালানোই সমাধান

হাজার মেশিনে প্রতিদিনই কিছু মেশিন বিকল হয়। MapReduce-এর উত্তর সরল: map আর reduce ফাংশন deterministic আর পার্শ্বপ্রতিক্রিয়াহীন হলে, বিকল মেশিনের task অন্য মেশিনে আবার চালালেই একই ফল আসবে। একজন master নিয়মিত ping করে বিকল worker চেনে। Map-এর output থাকে map মেশিনের local disk-এ, তাই মেশিন বিকল হলে শেষ হওয়া map task-ও আবার চালাতে হয়; reduce-এর output থাকে GFS-এ, তাই শেষ হওয়া reduce আর চালাতে হয় না। আর output ফাইল প্রথমে অস্থায়ী নামে লেখা হয়, শেষে atomic rename — তাই একই task দুবার চললেও চূড়ান্ত ফাইল একটিই। এই কৌশলটি পর্ব ৩৪-এর idempotent লেখার batch রূপ।

৪ · Dataflow engine, batch join আর ETL

MapReduce-এর পরে আসে dataflow engine — Apache Spark তার সবচেয়ে পরিচিত উদাহরণ (UC Berkeley-তে গবেষণা প্রকল্প হিসেবে শুরু; এর মূল ধারণা RDD নিয়ে প্রবন্ধ NSDI ২০১২-তে)। মূল পরিবর্তন তিনটি:

• পুরো কাজ একটি DAG: শুধু map আর reduce নয় — filter, join, group, sort-এর মতো অনেক operator, যারা একটি directed acyclic graph-এ জোড়া। Engine পুরো graph দেখে বুঝতে পারে কোন ধাপগুলো একসাথে চালানো যায়, আর কোথায় shuffle অবশ্যই লাগবে।
• মাঝের ফল memory-তে: ধাপের মাঝে সব ফল disk-এর file system-এ লেখার বদলে যতটা সম্ভব memory-তে বা local disk-এ রাখা। পুনরাবৃত্ত কাজে এতে বড় গতি আসে; তবে কতটা — সেটি ডেটার আকার, memory আর কাজের ধরনের ওপর নির্ভর করে, আর memory না কুলালে ডেটা disk-এ নামেই (spill).
• ব্যর্থতায় lineage: কোন ডেটা কোন ধাপে কী থেকে তৈরি — সেই রেসিপি মনে রাখা। একটি টুকরো হারালে শুধু সেই টুকরোটি রেসিপি মেনে আবার হিসাব করা হয়, পুরো কাজ নয়।

Spark-ধাঁচের engine DAG-কে stage-এ ভাগ করে: যে ধাপে প্রতিটি টুকরো শুধু নিজের ইনপুট দেখে (filter, map), সেগুলো এক stage-এ পাইপের মতো চলে; যে ধাপে অন্য টুকরোর ডেটা লাগে (group by key, join), সেখানে shuffle আর নতুন stage. তাই নকশা করার সময় প্রশ্নটি একই থাকে — কতবার shuffle হচ্ছে, আর প্রতিবার কত ডেটা?

Batch-এ join — sort-merge আর broadcast

ধরা যাক একটি কাল্পনিক ই-কমার্স সেবার ২ TB অর্ডার টেবিলকে ৫০ MB-র শহর টেবিলের সাথে join করতে হবে (city_id দিয়ে)। দুই কৌশল:

এখানে broadcast স্পষ্ট জয়ী: ১০০টি worker-এ ৫০ MB পাঠাতে মোট ৫ GB, আর sort-merge-এ ২ TB-র পুরোটাই নেটওয়ার্কে ঘুরত। কিন্তু ছোট টেবিল worker-এর memory ছাড়ালে broadcast ব্যর্থ হয়। অনেক engine টেবিলের আকার দেখে নিজেই বেছে নেয়; সেই সীমা configuration-এ থাকে আর সংস্করণভেদে আলাদা। আর join key-তে skew থাকলে (একটি শহরে অর্ধেক অর্ডার) sort-merge-এর একটি partition বিশাল হয় — অংশ ৩-এর সেই hot key.

ETL, ELT, data lake আর warehouse

Batch-এর সবচেয়ে বড় ব্যবহার বিশ্লেষণের ডেটা তৈরি। পুরোনো ধারা ETL: উৎস থেকে Extract, আলাদা সার্ভারে Transform (পরিষ্কার, যোগ, সারসংক্ষেপ), তারপর warehouse-এ Load. নতুন ধারা ELT: কাঁচা ডেটা আগে সস্তা storage-এ বা warehouse-এ তুলে ফেলা, transform পরে warehouse-এর নিজের শক্তিশালী engine-এ, SQL দিয়ে। কাঁচা ফাইলের সেই বিশাল ভান্ডারকে বলে data lake — সাধারণত object storage-এ (পর্ব ২৬), column-ভিত্তিক ফাইল ফরম্যাটে (column store, পর্ব ১৯)। আর সাজানো, schema-মানা, report-এর জন্য তৈরি অংশ হলো data warehouse. Star schema, slowly changing dimension আর incremental load-এর বিস্তারিত পাবেন DBMS কোর্সের data warehouse পর্বে।

৫ · Stream processing-এর মূল — event time আর window

এবার অসীম ধারায় আসি। Stream processor একটি log বা broker থেকে (পর্ব ৩১, পর্ব ৩৩) ঘটনা পড়ে, প্রতিটির ওপর হিসাব করে, আর ফল পাঠায় sink-এ — একটি database, cache, dashboard বা আরেকটি topic-এ। কিছু কাজে কোনো স্মৃতি লাগে না: “প্রতিটি পেমেন্ট ৫ লাখ টাকার বেশি হলে সতর্কতা” — প্রতিটি ঘটনা আলাদাভাবে দেখা যায়। কিন্তু “গত ১ মিনিটে একই কার্ডে ৫টির বেশি লেনদেন” বলতে গেলে সময় ধরে গুনতে হয়। আর সেখানেই প্রশ্ন: কোন সময়?

Event time বনাম processing time

একজন ব্যবহারকারী মেট্রোরেলে বসে অ্যাপে তিনটি বোতাম চাপলেন; সুড়ঙ্গে নেটওয়ার্ক নেই, তাই অ্যাপ ঘটনাগুলো জমিয়ে রাখল, আর ২ মিনিট পরে স্টেশনে পৌঁছে একসাথে পাঠাল। প্রতিটি ঘটনার দুটি সময়: ফোনে চাপার মুহূর্ত — event time, আর processor-এর হাতে পৌঁছানোর মুহূর্ত — processing time. নেটওয়ার্ক, retry, broker-এর জমা, processor-এর বিরতি — সব মিলিয়ে দুটির ফারাক মিলিসেকেন্ড থেকে ঘণ্টা, আর ঘটনাগুলো পৌঁছায় এলোমেলো ক্রমে।

Processing time ধরে গোনা সহজ — processor-এর নিজের ঘড়ি দেখলেই হলো। কিন্তু তখন “১২:০০–১২:০১-এ কতজন কার্ট-এ পণ্য রেখেছেন”-এর উত্তর নির্ভর করে নেটওয়ার্কের মেজাজের ওপর, আর একই ডেটা আবার চালালে উত্তর বদলায়। ব্যবসার প্রায় সব প্রশ্ন আসলে event time-এর প্রশ্ন। অংশ ৬-এর simulator-এ দেখবেন ফারাকটি কত বড়।

তিন রকম window

Sliding window-এর খরচ একটু হিসাব করুন। দৈর্ঘ্য ১০ মিনিট, slide ১ মিনিট, আর সেকেন্ডে ৫০,০০০ ঘটনা। প্রতিটি ঘটনা ১০ ÷ ১ = ১০টি window-এ পড়ে, তাই সাদামাটা বাস্তবায়নে সেকেন্ডে ৫০,০০০ × ১০ = ৫,০০,০০০টি state হালনাগাদ। চালাক বাস্তবায়ন সময়কে ১ মিনিটের টুকরোয় (slice বা pane) ভাগ করে: প্রতিটি ঘটনা একটিই টুকরোয় যোগ হয় (সেকেন্ডে ৫০,০০০ হালনাগাদ), আর প্রতি মিনিটে শেষ ১০টি টুকরো মিলিয়ে ফল বানানো হয়। শর্ত আবার সেই combiner-এর — হিসাবটি যোগ করা যায় এমন হতে হবে।

Session window ধারায় সবচেয়ে মজার, কারণ দেরিতে আসা একটি ঘটনা দুটি আলাদা session-কে জুড়ে দিতে পারে। নিচের কোডে দুজন ব্যবহারকারীর click, gap ৫ মিনিট। নিয়ম: একটি click আগের session-এর প্রথম click-এর ৫ মিনিট আগে থেকে শেষ click-এর ৫ মিনিট পর পর্যন্ত (দুই প্রান্তসহ) পড়লে সেই session-এ যোগ হয়; session-এর window হলো [প্রথম click, শেষ click + gap). রিনার একটি click মোবাইল offline থাকায় দেরিতে পৌঁছাচ্ছে।

// Session windows: a session ends after GAP minutes with no click from that user.
const GAP = 5;
const clicks = [                       // [user, event time in minutes], in arrival order
  ["rina", 0], ["tanvir", 1], ["rina", 2], ["rina", 3], ["tanvir", 9], ["tanvir", 10],
  ["tanvir", 11], ["rina", 12], ["rina", 14], ["tanvir", 25], ["rina", 30],
  ["rina", 7] ];                       // the last click was made offline and arrives late
const sessions = new Map();            // user -> list of [first click, last click, clicks]
function add(user, t) {
  const list = sessions.get(user) || [];
  let s = [t, t, 1];
  for (const o of list.filter((o) => t >= o[0] - GAP && t <= o[1] + GAP)) { // touches o?
    s = [Math.min(s[0], o[0]), Math.max(s[1], o[1]), s[2] + o[2]];
    list.splice(list.indexOf(o), 1);   // merge: the old session is absorbed
  }
  list.push(s); list.sort((a, b) => a[0] - b[0]); sessions.set(user, list);
}
const show = () => { for (const [u, list] of sessions)
  console.log(`  ${u.padEnd(7)}` + list.map(([a, b, n]) => `[${a}-${b + GAP}) ${n} clicks`).join("   ")); };
clicks.forEach(([u, t], i) => {
  add(u, t);
  if (i === clicks.length - 2) { console.log(`before the late click (gap ${GAP} min):`); show(); }
});
console.log(`after the late click at minute ${clicks.at(-1)[1]}:`); show();

before the late click (gap 5 min):
  rina   [0-8) 3 clicks   [12-19) 2 clicks   [30-35) 1 clicks
  tanvir [1-6) 1 clicks   [9-16) 3 clicks   [25-30) 1 clicks
after the late click at minute 7:
  rina   [0-19) 6 clicks   [30-35) 1 clicks
  tanvir [1-6) 1 clicks   [9-16) 3 clicks   [25-30) 1 clicks

৬ · Watermark আর দেরিতে আসা ডেটা

Event time ধরে ১২:০০–১২:০১ window গুনছি। কিন্তু কখন বলব “এই মিনিটের গণনা শেষ, এবার ফল পাঠানো যাক”? ১২:০১ বাজলেই নয় — ১২:০০:৫৯-এর একটি ঘটনা হয়তো এখনো মেট্রোর সুড়ঙ্গে। অনন্তকাল অপেক্ষাও করা যায় না। এখানে আসে watermark: event time-এর একটি চলমান রেখা, যার অর্থ “আমার ধারণা, এর চেয়ে পুরোনো event time-এর ঘটনা আর আসবে না।” Watermark কোনো window-এর শেষ সময় পেরোলে সেই window বন্ধ হয়, ফল বেরোয়।

সবচেয়ে প্রচলিত হিসাব: watermark = এ পর্যন্ত দেখা সবচেয়ে বড় event time − একটি নির্দিষ্ট দেরি (bounded out-of-orderness). দেরি বড় রাখলে ফল সম্পূর্ণতর, কিন্তু দেরিতে; ছোট রাখলে দ্রুত, কিন্তু কিছু ঘটনা বাদ। অংশ ১-এর latency বনাম completeness এখানে একটি সংখ্যায় নেমে এল। নিচের stepper-এ ১ মিনিটের tumbling window আর watermark = সর্বোচ্চ event time − ১০ সেকেন্ড:

এবার বড় পরিসরে চালান। নিচের simulator-এ ২৪টি click, প্রায় প্রতি ১০ সেকেন্ডে একটি, ৪ মিনিট ধরে — অর্থাৎ প্রতি মিনিটে ঠিক ৬টি। প্রায় ২০% click খুব দেরিতে পৌঁছায় (৬০–১১৯ সেকেন্ড), বাকিরা ০–১৪ সেকেন্ডে। Watermark-এর ছাড় ৩০ সেকেন্ড। একই ঘটনা তিনভাবে গোনা হয়েছে: আসল (event time, সব ঘটনা), watermark-সহ event time, আর processing time.

// Event-time tumbling windows (60 s) with a watermark, vs processing-time windows.
function rng(seed) {                                   // mulberry32: seeded, repeatable
  return () => { seed = (seed + 0x6D2B79F5) | 0; let t = Math.imul(seed ^ (seed >>> 15), 1 | seed);
    t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t; return ((t ^ (t >>> 14)) >>> 0) / 4294967296; };
}
const rand = rng(7), W = 60, DELAY = 30;              // watermark = largest event time seen - 30 s
const ev = [];
for (let i = 0; i < 24; i++) {                         // one click about every 10 s of event time
  const t = i * 10 + Math.floor(rand() * 10);
  const lag = rand() < 0.2 ? 60 + Math.floor(rand() * 60) : Math.floor(rand() * 15); // 20% very late
  ev.push({ id: "e" + i, t, arrive: t + lag });
}
ev.sort((a, b) => a.arrive - b.arrive || a.t - b.t);  // the processor sees arrival order
const fmt = (s) => `${Math.floor(s / 60)}:${String(s % 60).padStart(2, "0")}`;
const bump = (m, k) => m.set(k, (m.get(k) || 0) + 1);
const open = new Map(), emitted = new Map(), truth = new Map(), proc = new Map(), late = [];; let maxT = -Infinity;
for (const e of ev) {
  const start = Math.floor(e.t / W) * W;
  bump(truth, start); bump(proc, Math.floor(e.arrive / W) * W);
  if (emitted.has(start)) late.push(`${e.id}(t=${fmt(e.t)}, came ${fmt(e.arrive)})`);
  else bump(open, start);
  maxT = Math.max(maxT, e.t);
  const wm = maxT - DELAY;
  for (const s of [...open.keys()].sort((a, b) => a - b))
    if (s + W <= wm) {                                 // window end passed: emit final result
      console.log(`at ${fmt(e.arrive)} watermark ${fmt(wm)} closes [${fmt(s)}-${fmt(s + W)}) count=${open.get(s)}`);
      emitted.set(s, open.get(s)); open.delete(s);
    }
}
console.log("still open: " + [...open].map(([s, c]) => `[${fmt(s)}-${fmt(s + W)}) ${c}`).join(", "));
console.log("late, dropped: " + late.join(" "));
console.log("\nwindow        true  event-time  processing-time");
for (const s of [0, 60, 120, 180, 240])
  console.log(`[${fmt(s)}-${fmt(s + W)})`.padEnd(13) + String(truth.get(s) || 0).padStart(5) +
    String(emitted.get(s) ?? open.get(s) ?? 0).padStart(12) + String(proc.get(s) || 0).padStart(17));

at 1:55 watermark 1:15 closes [0:00-1:00) count=5
at 2:48 watermark 2:04 closes [1:00-2:00) count=5
at 3:50 watermark 3:11 closes [2:00-3:00) count=5
still open: [3:00-4:00) 6
late, dropped: e0(t=0:00, came 1:58) e9(t=1:37, came 3:31) e17(t=2:53, came 4:00)

window        true  event-time  processing-time
[0:00-1:00)      6           5                4
[1:00-2:00)      6           5                5
[2:00-3:00)      6           5                6
[3:00-4:00)      6           6                7
[4:00-5:00)      0           0                2

দেরির ডেটা — allowed lateness আর আলাদা পথ

Window বন্ধের পরে আসা ঘটনাকে বলে late data. তিনটি বিকল্প: ফেলে দেওয়া (simulator তাই করেছে); বন্ধ window-এর state কিছুক্ষণ রেখে ফল সংশোধন করা — allowed lateness; আর সীমার পরেও এলে আলাদা একটি ধারায় পাঠানো (side output), যাতে পরে batch-এ মিলিয়ে নেওয়া যায় বা অন্তত গোনা যায় কতগুলো হারাল। নিচের stepper-এ আগের অবস্থা থেকে শুরু, allowed lateness ৩০ সেকেন্ড:

আরেকটি বাস্তব ফাঁদ: নিষ্ক্রিয় উৎস। বহু partition থেকে পড়লে processor-এর watermark হয় সব partition-এর watermark-এর সর্বনিম্ন। রাতে একটি partition-এ কোনো ঘটনা না এলে তার watermark থেমে থাকে, আর পুরো processor-এর সব window আটকে যায়। অনেক engine-এ তাই “idle” partition চিহ্নিত করে বাদ রাখার ব্যবস্থা থাকে; নকশায় এটি মাথায় রাখতে হয়।

৭ · State, checkpoint, stream join আর exactly-once

Window-এর গণনা, session-এর তালিকা, “এই কার্ডে গত ১ মিনিটে কটি লেনদেন” — এসবই stream processor-এর state. State থাকে key অনুযায়ী ভাগ করা (keyed state): একই কার্ডের সব ঘটনা একই processor instance-এ যায় (log-এর partition key দিয়ে, পর্ব ৩৩), তাই state স্থানীয় memory বা local disk-এ রাখা যায়, প্রতিটি ঘটনায় দূরের database-এ যেতে হয় না। বড় state-এর জন্য অনেক engine ভেতরে একটি LSM-ধাঁচের embedded store ব্যবহার করে (পর্ব ১৯)।

Checkpoint — crash-এর পরে কোথা থেকে শুরু

Processor crash করলে memory-র state হারায়। সমাধান নিয়মিত checkpoint: একটি নির্দিষ্ট মুহূর্তে সব operator-এর state আর ইনপুট log-এর কোন offset পর্যন্ত পড়া হয়েছে — দুটোই একসাথে টেকসই storage-এ লেখা। Crash-এর পরে শেষ checkpoint-এর state ফিরিয়ে আনা হয়, আর log-কে সেই offset-এ ফিরিয়ে আবার পড়া শুরু হয়। ফলে checkpoint-এর পরের ঘটনাগুলো দ্বিতীয়বার প্রক্রিয়া হয়, কিন্তু state-ও সেই মুহূর্তে ফিরে গেছে — তাই state-এর হিসাবে প্রতিটি ঘটনার প্রভাব ঠিক একবার। এর জন্য উৎসকে replay করা যায় এমন হতে হবে; সাধারণ queue থেকে পড়া বার্তা মুছে গেলে (পর্ব ৩১) এই কৌশল চলে না, log চলে।

কিন্তু শত শত operator-এর state “একই মুহূর্তে” কীভাবে? সব বন্ধ করে ছবি তুললে ধারা থেমে যাবে। Flink-ধাঁচের engine-এ উৎস থেকে ধারার ভেতরে একটি বিশেষ চিহ্ন (barrier) পাঠানো হয়; প্রতিটি operator barrier পেলে নিজের state-এর ছবি তোলে আর barrier সামনে এগিয়ে দেয়। এটি Chandy–Lamport-এর distributed snapshot (১৯৮৫) ভাবনার একটি রূপ; Flink দলের প্রকাশিত বর্ণনা (Carbone ও সহকর্মীরা, ২০১৫) একে asynchronous barrier snapshotting বলে।

Exactly-once — কোন অর্থে

Engine-এর বিজ্ঞাপনে “exactly-once” শব্দটি খুব দেখা যায়। এর সৎ অর্থ: state আর output-এর ওপর প্রতিটি ঘটনার প্রভাব একবার — ঘটনা আসলে একাধিকবার প্রক্রিয়া হলেও। ভেতরের state-এর জন্য checkpoint যথেষ্ট। বাইরের sink-এর জন্য দুটির একটি লাগে: sink idempotent (window key ধরে upsert, যেমন উপরের stepper-এ), অথবা output checkpoint-এর সাথে transaction-এ বাঁধা (Kafka-র transaction, লেখার সময়ে Kafka ০.১১ থেকে, ২০১৭)। আর SMS বা email পাঠানোর মতো ফেরানো যায় না এমন কাজে “exactly-once” প্রায় কখনো সম্ভব নয় — সেখানে idempotency key আর deduplication. পুরো আলোচনা পর্ব ৩৪-এ।

Stream join — ঘটনার সাথে টেবিল মেলানো

প্রতিটি অর্ডার-ঘটনায় শুধু user_id আছে, কিন্তু dashboard-এ চাই ব্যবহারকারীর শহর। প্রতিটি ঘটনায় database-এ query করলে সেকেন্ডে লাখো query — database ভেঙে পড়বে, আর latency বাড়বে। ভালো পথ stream-table join (enrichment): user টেবিলের প্রতিটি বদল CDC-এর মাধ্যমে একটি log-এ আসে; processor সেই log পড়ে টেবিলটির একটি স্থানীয় কপি নিজের state-এ রাখে, আর প্রতিটি অর্ডার এলে স্থানীয়ভাবে মেলায়। Kafka Streams-এ এটি KStream-KTable join নামে পরিচিত; Flink-এ lookup বা temporal join (নাম আর খুঁটিনাটি সংস্করণভেদে আলাদা)। সূক্ষ্ম প্রশ্ন: অর্ডারটি মেলানো হবে ঘটনার সময়ের শহরের সাথে, নাকি এখনকার শহরের সাথে? Event-time-ভিত্তিক temporal join প্রথমটি দেয়; সাধারণ lookup দ্বিতীয়টি।

আরেক রকম stream-stream join: দুটি ধারার ঘটনা মেলানো, যেমন “বিজ্ঞাপন দেখানো” আর “বিজ্ঞাপনে click”. দুটি ধারাই অসীম, তাই একটি সময়সীমা লাগে — “দেখানোর ৩০ মিনিটের মধ্যে click” — নইলে state অসীম হয়। সীমা পেরোলে watermark দেখে পুরোনো ঘটনা state থেকে মোছা হয়।

৮ · Lambda বনাম kappa — দুই পথ, নাকি এক পথ

২০১০-এর দশকের শুরুতে stream engine-গুলো দ্রুত হলেও নির্ভরযোগ্য ছিল না — ঘটনা হারাত বা দুবার গুনত, আর দেরির ডেটা সামলাত না। Nathan Marz (Apache Storm-এর স্রষ্টা) তখন একটি নকশা প্রস্তাব করেন, যার নাম lambda architecture (২০১১ সালের একটি blog লেখা থেকে শুরু; পরে James Warren-এর সাথে লেখা “Big Data” বইয়ে, ২০১৫)। ভাবনাটি: সব কাঁচা ঘটনা অপরিবর্তনীয়ভাবে জমা রাখুন; একটি batch layer পুরো ইতিহাস থেকে নির্ভুল ফল বানাক; একটি speed layer শুধু সাম্প্রতিক ঘটনা থেকে দ্রুত আনুমানিক ফল দিক; আর query দুটো মিলিয়ে উত্তর দিক। পরের batch চললে speed layer-এর ভুল আপনিই মুছে যায়।

সমস্যা: একই ব্যবসার নিয়ম দুবার লিখতে হয় — দুটি ভিন্ন system-এ, প্রায়ই দুটি ভিন্ন API-তে — আর দুটিকে সবসময় মিলিয়ে রাখতে হয়। ২০১৪ সালে LinkedIn-এ Kafka তৈরিতে জড়িত Jay Kreps “Questioning the Lambda Architecture” লেখায় বিকল্প দেন, পরে যার নাম হয় kappa architecture: শুধু একটি stream পথ। সব ঘটনা যথেষ্ট দীর্ঘ retention-সহ log-এ রাখুন। Logic বদলালে নতুন সংস্করণের job log-এর শুরু থেকে আবার চালান, নতুন output টেবিলে লিখুন; সে বর্তমানে ধরে ফেললে পাঠকদের নতুন টেবিলে সরিয়ে পুরোনো job বন্ধ করুন।

আজকের বাস্তবতা প্রায়ই মাঝামাঝি: একই engine বা একই SQL দিয়ে stream আর batch দুটোই লেখা, যাতে নিয়ম একবার লিখতে হয়। নাম মুখস্থের চেয়ে জরুরি প্রশ্ন: “logic-এ bug পেলে গত তিন মাসের ফল কীভাবে ঠিক করব?” এই প্রশ্নের উত্তর না থাকলে নকশা অসম্পূর্ণ।

৯ · AI-এর যুগে batch আর stream নকশা যাচাই

AI-এর কাছে একটি real-time analytics pipeline-এর নকশা চাইলে সে কয়েক সেকেন্ডেই সুন্দর একটি ছবি দেবে: Kafka, Flink, Spark, একটি data lake, একটি warehouse, সাথে Redis. কিন্তু দিনে ২০,০০০ ঘটনা হলে এই পুরো বহর চালানোর খরচ আর জটিলতা কাজের চেয়ে বহু গুণ বেশি; একটি database টেবিল আর প্রতি ৫ মিনিটের একটি SQL job হয়তো যথেষ্ট। AI সাধারণত যেসব জায়গায় ভুল করে:

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

• ঘটনার হার, আকার আর প্রয়োজনীয় latency নিজে হিসাব করেছি — stream আদৌ দরকার কি না, নাকি প্রতি ঘণ্টার batch যথেষ্ট
• প্রতিটি window-এ event time ব্যবহার হচ্ছে, timestamp কোথা থেকে আসছে আর watermark-এর ছাড় কত — লিখে নিয়েছি
• দেরির ডেটা, idle partition আর hot key দিয়ে ছোট simulation বা test চালিয়েছি
• প্রতিটি sink কীভাবে একই ফল দুবার পেলে সামলায় (upsert, idempotency key, transaction) — মিলিয়েছি
• Crash, consumer lag আর logic-এ bug — তিনটির পর পুনরুদ্ধারের পথ ধাপে ধাপে লিখেছি
• সব product-এর সীমা আর ডিফল্ট official documentation-এ মিলিয়েছি, AI-এর কথায় নয়