فصل ۱۰ — پردازش دسته‌ای

یک سیستم نمی‌تواند موفق باشد اگر بیش از حد تحت تأثیر یک نفر قرار گیرد. وقتی طراحی اولیه کامل و نسبتاً مقاوم شد، آزمایش واقعی آغاز می‌شود؛ افرادی با دیدگاه‌های بسیار متفاوت آزمایش‌های خود را انجام می‌دهند. — Donald Knuth

در دو بخش اول این کتاب زیاد درباره درخواست‌ها و کوئری‌ها و پاسخ‌ها یا نتایج متناظرشان صحبت کردیم. این سبک پردازش داده در بسیاری از سیستم‌های داده مدرن فرض شده است: چیزی می‌خواهید یا دستوری می‌فرستید، و مدتی بعد سیستم (امیدواریم) پاسخ می‌دهد. پایگاه‌های داده، cacheها، ایندکس‌های جستجو، وب‌سرورها و بسیاری سیستم‌های دیگر این‌طور کار می‌کنند.

در چنین سیستم‌های آنلاینی، چه مرورگر وب صفحه‌ای درخواست کند یا سرویسی API راه‌دور را فراخوانی کند، معمولاً فرض می‌کنیم درخواست توسط کاربر انسانی راه می‌افتد و کاربر منتظر پاسخ است. نباید زیاد منتظر بماند، پس به زمان پاسخ این سیستم‌ها توجه زیادی می‌دهیم (به «توصیف عملکرد» در صفحه ۱۳ مراجعه کنید).

وب و APIهای فزاینده مبتنی بر HTTP/REST، سبک تعامل درخواست/پاسخ را آن‌قدر رایج کرده که گرفتن آن به‌عنوان مسلم آسان است. اما به یاد داشته باشیم تنها راه ساخت سیستم نیست و رویکردهای دیگر هم مزیت دارند. سه نوع سیستم را از هم جدا کنیم:

سرویس‌ها (سیستم‌های آنلاین) یک سرویس منتظر می‌ماند تا درخواست یا دستوری از کلاینت برسد. وقتی دریافت شد، سعی می‌کند هرچه سریع‌تر آن را پردازش کند و پاسخ برگرداند. زمان پاسخ معمولاً معیار اصلی عملکرد سرویس است و در دسترس بودن اغلب بسیار مهم است (اگر کلاینت به سرویس دسترسی نداشته باشد، احتمالاً پیام خطا می‌گیرد).

سیستم‌های پردازش دسته‌ای (سیستم‌های آفلاین) یک سیستم پردازش دسته‌ای مقدار زیادی داده ورودی می‌گیرد، jobای برای پردازش آن اجرا می‌کند و داده خروجی تولید می‌کند. jobها اغلب مدتی طول می‌کشند (از چند دقیقه تا چند روز)، پس معمولاً کاربری منتظر پایان job نیست. در عوض، jobهای دسته‌ای اغلب دوره‌ای زمان‌بندی می‌شوند (مثلاً روزی یک‌بار). معیار اصلی عملکرد job دسته‌ای معمولاً throughput است (زمان لازم برای پردازش مجموعه ورودی با اندازه مشخص). پردازش دسته‌ای را در این فصل بررسی می‌کنیم.

سیستم‌های پردازش جریانی (سیستم‌های نزدیک به real-time) پردازش جریانی جایی بین آنلاین و آفلاین/دسته‌ای است (گاه near-real-time یا nearline هم می‌گویند). مثل سیستم دسته‌ای، پردازشگر جریانی ورودی می‌خورد و خروجی تولید می‌کند (نه پاسخ به درخواست). اما job جریانی کمی بعد از وقوع رویدادها روی آن‌ها کار می‌کند، در حالی که job دسته‌ای روی مجموعه ثابتی از داده ورودی کار می‌کند. این تفاوت به سیستم‌های جریانی اجازه می‌دهد latency کمتری نسبت به معادل دسته‌ای داشته باشند. چون پردازش جریانی بر پایه پردازش دسته‌ای ساخته می‌شود، در فصل ۱۱ بحث می‌کنیم.

همان‌طور که در این فصل می‌بینیم، پردازش دسته‌ای سنگ بنای مهمی در ساخت برنامه‌های قابل اعتماد، مقیاس‌پذیر و قابل نگهداری است. مثلاً MapReduce، الگوریتم پردازش دسته‌ای منتشرشده در ۲۰۰۴ [۱]، (شاید با اشتیاق زیاد) «الگوریتمی که Google را massively scalable کرد» [۲] نامیده شد. بعداً در سیستم‌های داده متن‌باز مختلف از جمله Hadoop، CouchDB و MongoDB پیاده‌سازی شد.

MapReduce نسبت به سیستم‌های پردازش موازی که سال‌ها قبل برای انبارهای داده توسعه یافته بودند [۳، ۴] مدل برنامه‌نویسی نسبتاً سطح پایین‌تری است، اما گام بزرگی در مقیاس پردازش روی سخت‌افزار commodity بود. اگرچه اهمیت MapReduce اکنون کاهش یافته [۵]، فهمیدنش ارزش دارد چون تصویر روشنی از چرایی و چگونگی مفید بودن پردازش دسته‌ای می‌دهد.

در واقع، پردازش دسته‌ای شکل بسیار قدیمی محاسبه است. پیش از اختراع رایانه‌های دیجیتال برنامه‌پذیر، ماشین‌های tabulating با کارت سوراخ‌دار — مثل ماشین‌های Hollerith در سرشماری ۱۸۹۰ آمریکا [۶] — شکل نیمه‌مکانیکی پردازش دسته‌ای را برای محاسبه آمار تجمیعی از ورودی بزرگ پیاده کردند. و MapReduce شباهت عجیبی به ماشین‌های مرتب‌سازی کارت electromechanical IBM دارد که در دهه ۱۹۴۰ و ۱۹۵۰ برای پردازش داده تجاری رایج بودند [۷]. مثل همیشه، تاریخ تمایل دارد تکرار شود.

در این فصل MapReduce و چند الگوریتم و فریم‌ورک پردازش دسته‌ای دیگر را می‌بینیم و بررسی می‌کنیم در سیستم‌های داده مدرن چگونه استفاده می‌شوند. اما برای شروع، پردازش داده با ابزارهای استاندارد Unix را می‌بینیم. حتی اگر با آن‌ها آشنا باشید، یادآوری فلسفه Unix ارزش دارد چون ایده‌ها و درس‌های Unix به سیستم‌های داده توزیع‌شده ناهمگن در مقیاس بزرگ منتقل می‌شوند.

پردازش دسته‌ای با ابزارهای Unix

با مثال ساده شروع کنیم. فرض کنید وب‌سروری دارید که با هر درخواست سرویس‌دهی، خطی به فایل log اضافه می‌کند. مثلاً با فرمت پیش‌فرض access log در nginx، یک خط log شبیه این است:

216.58.210.78 - - [27/Feb/2015:17:55:11 +0000] "GET /css/typography.css HTTP/1.1"
200 3377 "http://martin.kleppmann.com/" "Mozilla/5.0 (Macintosh; Intel Mac OS X
10_9_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/40.0.2214.115
Safari/537.36"

(در واقع یک خط است؛ فقط برای خوانایی به چند خط شکسته شده.)

اطلاعات زیادی در آن خط است. برای تفسیر باید تعریف فرمت log را ببینید:

$remote_addr - $remote_user [$time_local] "$request"
$status $body_bytes_sent "$http_referer" "$http_user_agent"

پس این یک خط log نشان می‌دهد در ۲۷ فوریه ۲۰۱۵، ساعت ۱۷:۵۵:۱۱ UTC، سرور درخواست فایل /css/typography.css را از IP کلاینت 216.58.210.78 دریافت کرد. کاربر احراز هویت نشده بود، پس $remote_user خط تیره (-) است. وضعیت پاسخ ۲۰۰ بود (یعنی درخواست موفق) و پاسخ ۳٬۳۷۷ بایت بود. مرورگر Chrome 40 بود و فایل را چون در صفحه http://martin.kleppmann.com/ ارجاع داده شده بود بارگذاری کرد.

تحلیل ساده log

ابزارهای مختلف می‌توانند این فایل‌های log را بگیرند و گزارش‌های زیبایی از ترافیک وب‌سایت بسازند، اما برای تمرین، با ابزارهای پایه Unix خودمان بسازیم. مثلاً می‌خواهید پنج صفحه محبوب‌تر وب‌سایت را پیدا کنید. در shell Unix این‌طور می‌توانید:i

cat /var/log/nginx/access.log |
  awk '{print $7}' |
  sort             |
  uniq -c          |
  sort -r -n       |
  head -n 5

• فایل log را می‌خواند.
• هر خط را با فاصله به فیلد تقسیم می‌کند و فقط فیلد هفتم (URL درخواستی) را چاپ می‌کند. در خط مثال، URL درخواست /css/typography.css است.
• لیست URLها را الفبایی مرتب می‌کند. اگر URLی n بار درخواست شده باشد، بعد از مرتب‌سازی همان URL n بار پشت سر هم تکرار می‌شود.
• دستور uniq خطوط تکراری مجاور را حذف می‌کند. گزینه -c شمارنده هم می‌دهد: برای هر URL متمایز، چند بار در ورودی آمده گزارش می‌کند.
• مرتب‌سازی دوم بر اساس عدد (-n) ابتدای هر خط (تعداد درخواست URL) است و نتیجه را معکوس (-r) برمی‌گرداند، یعنی بیشترین اول.
• در نهایت head فقط پنج خط اول (-n 5) را چاپ می‌کند.

خروجی آن زنجیره دستورات شبیه این است:

4189 /favicon.ico
3631 /2013/05/24/improving-security-of-ssh-private-keys.html
2124 /2012/12/05/schema-evolution-in-avro-protocol-buffers-thrift.html
1369 /
 915 /css/typography.css

اگر با ابزارهای Unix آشنا نیستید، خط فرمان بالا ممکن است مبهم به نظر برسد، اما فوق‌العاده قدرتمند است. گیگابایت‌ها log را در چند ثانیه پردازش می‌کند و تحلیل را به‌راحتی تغییر می‌دهید. مثلاً برای حذف فایل‌های CSS از گزارش، آرگومان awk را به '$7 !~ /\.css$/ {print $7}' تغییر دهید. برای شمارش IPهای برتر به‌جای صفحات، awk را به '{print $1}' تغییر دهید. و غیره.

فضای کتاب برای جزئیات ابزارهای Unix نیست، اما یادگیری‌شان ارزش دارد. شگفت‌آور تعداد زیادی تحلیل داده با ترکیبی از awk، sed، grep، sort، uniq و xargs در چند دقیقه انجام می‌شود و عملکردشان خوب است [۸].

i. بعضی‌ها دوست دارند بگویند cat اینجا لازم نیست و می‌توان فایل ورودی را مستقیم به awk داد. اما pipeline خطی وقتی این‌طور نوشته شود واضح‌تر است.

زنجیره دستورات در برابر برنامه سفارشی

به‌جای زنجیره دستورات Unix می‌توانید برنامه ساده‌ای بنویسید. مثلاً در Ruby شبیه این است:

counts = Hash.new(0)

File.open('/var/log/nginx/access.log') do |file|
  file.each do |line|
    url = line.split[6]
    counts[url] += 1
  end
end

top5 = counts.map{|url, count| [count, url] }.sort.reverse[0...5]
top5.each{|count, url| puts "#{count} #{url}" }

• counts جدول hash است که برای هر URL شمارنده نگه می‌دارد. پیش‌فرض صفر است.
• از هر خط log، URL فیلد هفتم جدا شده با فاصله است (ایندکس ۶ چون آرایه‌های Ruby از صفر شروع می‌شوند).
• شمارنده URL در خط فعلی log افزایش می‌یابد.
• محتوای hash بر اساس شمارنده (نزولی) مرتب و پنج مورد برتر گرفته می‌شود.
• پنج مورد برتر چاپ می‌شوند.

این برنامه به‌اندازه pipeline Unix مختصر نیست، اما نسبتاً خواناست و ترجیح شما تا حدی سلیقه است. اما علاوه بر تفاوت‌های سطحی نحوی، تفاوت بزرگی در جریان اجرا هست که روی فایل بزرگ آشکار می‌شود.

مرتب‌سازی در برابر تجمیع در حافظه

اسکریپت Ruby جدول hash URLها را در حافظه نگه می‌دارد و هر URL به تعداد دفعات دیده‌شدن map شده. مثال pipeline Unix چنین hash tableای ندارد و به مرتب‌سازی لیست URLها تکیه می‌کند که تکرارهای یک URL ساده تکرار شده‌اند.

کدام بهتر است؟ به تعداد URLهای متمایز بستگی دارد. برای اکثر وب‌سایت‌های کوچک تا متوسط، احتمالاً همه URLهای متمایز و شمارنده هر کدام در (مثلاً) ۱ گیگابایت حافظه جا می‌شوند. در این مثال، working set job (حافظه‌ای که job به دسترسی تصادفی نیاز دارد) فقط به تعداد URLهای متمایز بستگی دارد: اگر یک میلیون ورودی log برای یک URL باشد، فضای hash table هنوز فقط یک URL به‌علاوه اندازه شمارنده است. اگر working set کوچک باشد، hash table در حافظه خوب کار می‌کند — حتی روی لپ‌تاپ.

از طرف دیگر، اگر working set از حافظه موجود بزرگ‌تر باشد، رویکرد مرتب‌سازی مزیت دارد که از دیسک به‌کارآمد استفاده می‌کند. همان اصل «SSTables و LSM-Trees» در صفحه ۷۶: قطعه‌های داده در حافظه مرتب و به‌عنوان segment روی دیسک نوشته می‌شوند و segmentهای مرتب چندتایی merge می‌شوند. Mergesort الگوی دسترسی sequential دارد که روی دیسک خوب عمل می‌کند. (بهینه‌سازی برای I/O sequential موضوع تکراری فصل ۳ بود. همان الگو اینجا هم هست.)

ابزار sort در GNU Coreutils (Linux) با spill به دیسک datasetهای بزرگ‌تر از حافظه را خودکار مدیریت می‌کند و مرتب‌سازی را روی چند هسته CPU موازی می‌کند [۹]. پس زنجیره ساده Unix که دیدیم به‌راحتی به dataset بزرگ scale می‌کند بدون تمام شدن حافظه. گلوگاه احتمالاً نرخ خواندن فایل ورودی از دیسک است.

فلسفه Unix

اتفاقی نیست که تحلیل log را با زنجیره دستورات مثل مثال قبل به‌راحتی انجام دادیم: این یکی از ایده‌های کلیدی طراحی Unix بود و امروز هم شگفت‌آور مرتبط است. عمیق‌تر نگاه کنیم تا ایده‌هایی از Unix قرض بگیریم [۱۰].

Doug McIlroy، مخترع pipe در Unix، در ۱۹۶۴ این‌طور توصیف کرد [۱۱]: «باید راهی برای وصل کردن برنامه‌ها مثل شلنگ باغ داشته باشیم — وقتی لازم شد قطعه دیگری بپیچیم تا داده را جور دیگری پردازش کنیم. این روش I/O هم هست.» تشبیه لوله‌کشی ماند و ایده وصل کردن برنامه‌ها با pipe بخشی از فلسفه Unix شد — مجموعه اصول طراحی که بین توسعه‌دهندگان و کاربران Unix محبوب شد. فلسفه در ۱۹۷۸ این‌طور توصیف شد [۱۲، ۱۳]:

1. هر برنامه یک کار را خوب انجام دهد. برای کار جدید، از نو بسازید نه اینکه با «ویژگی» جدید برنامه قدیمی را پیچیده کنید.
2. انتظار داشته باشید خروجی هر برنامه ورودی برنامه دیگری شود که هنوز نمی‌دانید چیست. خروجی را با اطلاعات اضافی شلوغ نکنید. از فرمت ورودی ستونی سخت یا باینری پرهیز کنید. ورودی تعاملی را اصرار نکنید.
3. نرم‌افزار و حتی سیستم‌عامل را زود امتحان کنید، ایده‌آل ظرف چند هفته. در دور انداختن بخش‌های ناجور و بازسازی تردید نکنید.
4. ابزار را به کمک ناشیانه ترجیح دهید تا کار برنامه‌نویسی سبک شود، حتی اگر برای ساخت ابزار منحرف شوید و بعد از استفاده بعضی را دور بیندازید.

این رویکرد — خودکارسازی، نمونه‌سازی سریع، تکرار تدریجی، دوستی با آزمایش، و شکستن پروژه بزرگ به قطعات قابل مدیریت — شبیه جنبش‌های Agile و DevOps امروز است. در چهار دهه شگفت‌آور کم تغییر کرده.

ابزار sort نمونه خوب برنامه‌ای است که یک کار را خوب انجام می‌دهد. می‌توان گفت پیاده‌سازی مرتب‌سازی بهتری نسبت به کتابخانه استاندارد اکثر زبان‌ها دارد (که به دیسک spill نمی‌کند و حتی وقتی مفید است از چند thread استفاده نمی‌کند). با این حال sort به‌تنهایی کم‌کاربرد است. فقط با ابزارهای دیگر Unix مثل uniq قدرتمند می‌شود.

Shell Unix مثل bash به‌راحتی این برنامه‌های کوچک را به jobهای پردازش داده قدرتمند وصل می‌کند. با وجود اینکه بسیاری از این برنامه‌ها توسط گروه‌های مختلف نوشته شده‌اند، به‌شکل انعطاف‌پذیر به هم وصل می‌شوند. Unix برای این composability چه می‌کند؟

رابط یکنواخت

اگر انتظار دارید خروجی یک برنامه ورودی برنامه دیگر شود، یعنی برنامه‌ها باید همان فرمت داده — یعنی رابط سازگار — را استفاده کنند. اگر بخواهید خروجی هر برنامه به ورودی هر برنامه دیگر وصل شود، همه باید همان رابط ورودی/خروجی را داشته باشند. در Unix این رابط فایل است (دقیق‌تر، file descriptor). فایل فقط دنباله‌ای مرتب از بایت است. چون رابط ساده است، چیزهای مختلف با همان رابط نمایش داده می‌شوند: فایل واقعی روی filesystem، کانال ارتباط با process دیگر (Unix socket، stdin، stdout)، درایور دستگاه (مثل /dev/audio یا /dev/lp0)، socket نشان‌دهنده اتصال TCP و غیره. گرفتن این به‌عنوان مسلم آسان است، اما شگفت‌آور است این چیزهای بسیار متفاوت رابط یکنواخت مشترک دارند و به‌راحتی به هم وصل می‌شوند.ii

بر اساس قرارداد، بسیاری (نه همه) برنامه‌های Unix این دنباله بایت را متن ASCII می‌گیرند. مثال تحلیل log از این استفاده کرد: awk، sort، uniq و head ورودی را لیست رکوردهای جدا شده با \n (newline، ASCII 0x0A) می‌گیرند. انتخاب \n دلخواه است — جداکننده رکورد ASCII 0x1E شاید بهتر بود چون برای این کار است [۱۴] — اما اینکه همه این برنامه‌ها روی همان جداکننده رکورد استاندارد شده‌اند باعث interoperate شدن می‌شود.

ii. نمونه دیگر رابط یکنواخت URL و HTTP، بنیاد وب است. URL چیز مشخصی (resource) در وب‌سایت را شناسایی می‌کند و می‌توانید از هر وب‌سایت دیگر به هر URL لینک دهید. کاربر با مرورگر وب بدون درز بین وب‌سایت‌ها با دنبال کردن لینک جابه‌جا می‌شود، حتی اگر سرورها توسط سازمان‌های بی‌ربط اداره شوند. امروز بدیهی به نظر می‌رسد، اما بینش کلیدی موفقیت وب بود. سیستم‌های قبلی یکنواخت نبودند: مثلاً در عصر BBS هر سیستم شماره تلفن و baud rate خودش را داشت. ارجاع از یک BBS به BBS دیگر باید شماره تلفن و تنظیمات مودم باشد؛ کاربر باید قطع کند، BBS دیگر را بگیرد و دستی اطلاعات را پیدا کند. لینک مستقیم به محتوای داخل BBS دیگر ممکن نبود.

تجزیه هر رکورد (خط ورودی) مبهم‌تر است. ابزارهای Unix معمولاً خط را با فاصله یا tab به فیلد تقسیم می‌کنند، اما CSV، pipe-separated و encodingهای دیگر هم هست. حتی ابزار ساده مثل xargs نیم‌ده گزینه خط فرمان برای نحوه parse ورودی دارد.

رابط یکنواخت متن ASCII بیشتر کار می‌کند، اما زیبا نیست: مثال log از {print $7} برای استخراج URL استفاده کرد که خوانا نیست. در دنیای ایده‌آل شاید {print $request_url} بود. بعداً برمی‌گردیم.

اگرچه کامل نیست، دهه‌ها بعد رابط یکنواخت Unix هنوز شگفت‌انگیز است. نرم‌افزار زیادی مثل ابزارهای Unix interoperate و compose نمی‌کند: نمی‌توانید راحت محتوای ایمیل و تاریخچه خرید آنلاین را از ابزار تحلیل سفارشی به spreadsheet بفرستید و نتیجه را در شبکه اجتماعی یا wiki بگذارید. امروز استثناست نه قاعده که برنامه‌ها به‌قدر ابزارهای Unix نرم با هم کار کنند.

حتی پایگاه‌های داده با مدل داده یکسان معمولاً بیرون آوردن داده از یکی و فرستادن به دیگری را آسان نمی‌کنند. این کمبود یکپارچگی به Balkanization داده منجر می‌شود.

جداسازی منطق و wiring

ویژگی دیگر ابزارهای Unix استفاده از standard input (stdin) و standard output (stdout) است. اگر برنامه‌ای اجرا کنید و چیز دیگری مشخص نکنید، stdin از صفحه‌کلید و stdout به صفحه می‌رود. اما می‌توانید ورودی از فایل بگیرید و/یا خروجی را به فایل redirect کنید. Pipeها stdout یک process را به stdin process دیگر وصل می‌کنند (با buffer کوچک در حافظه، بدون نوشتن کل جریان میانی روی دیسک).

برنامه هنوز می‌تواند مستقیم فایل بخواند/بنویسد، اما رویکرد Unix وقتی بهتر است برنامه نگران مسیر فایل خاص نباشد و فقط stdin و stdout استفاده کند. کاربر shell ورودی/خروجی را هرطور بخواهد وصل می‌کند؛ برنامه نمی‌داند ورودی از کجا و خروجی به کجا می‌رود. (می‌توان گفت loose coupling، late binding [۱۵] یا inversion of control [۱۶] است.) جدا کردن wiring ورودی/خروجی از منطق برنامه، compose کردن ابزار کوچک به سیستم بزرگ‌تر را آسان‌تر می‌کند.

می‌توانید برنامه خودتان را بنویسید و با ابزارهای سیستم‌عامل ترکیب کنید. برنامه فقط از stdin بخواند و به stdout بنویسد تا در pipeline پردازش داده شرکت کند. در مثال log می‌توانید ابزاری بنویسید که user-agent را به شناسه مرورگر معقول‌تر تبدیل کند، یا IP را به کد کشور، و در pipeline بگذارید. sort مهم نیست با بخش دیگر سیستم‌عامل صحبت می‌کند یا برنامه شما.

با این حال برای stdin و stdout حدودی هست. برنامه با چند ورودی/خروجی ممکن است اما پیچیده. نمی‌توانید خروجی برنامه را به اتصال شبکه pipe کنید [۱۷، ۱۸].iii اگر برنامه مستقیم فایل باز کند، subprocess راه بیندازد یا اتصال شبکه باز کند، I/O توسط خود برنامه wired up می‌شود. هنوز قابل پیکربندی است (مثلاً با گزینه خط فرمان)، اما انعطاف wiring در shell کمتر می‌شود.

iii. مگر با ابزار جدا مثل netcat یا curl. Unix ابتدا همه چیز را فایل می‌خواست، اما BSD sockets API از آن convention منحرف شد [۱۷]. Plan 9 و Inferno یکنواخت‌تر از فایل استفاده می‌کنند: اتصال TCP را فایلی در /net/tcp نشان می‌دهند [۱۸].

شفافیت و آزمایش

بخشی از موفقیت ابزارهای Unix این است که دیدن اتفاقات را آسان می‌کنند:

• فایل‌های ورودی دستورات Unix معمولاً immutable در نظر گرفته می‌شوند. یعنی می‌توانید دستورات را هرچند بار بخواهید با گزینه‌های مختلف اجرا کنید بدون آسیب به فایل ورودی.
• می‌توانید pipeline را هرجا قطع کنید، خروجی را به less pipe کنید و ببینید فرم مورد انتظار را دارد. این برای debug عالی است.
• می‌توانید خروجی یک مرحله pipeline را در فایل بنویسید و به‌عنوان ورودی مرحله بعد استفاده کنید. مرحله بعد را بدون اجرای کل pipeline از نو راه می‌اندازید.

پس حتی اگر ابزارهای Unix نسبت به optimizer کوئری relational ساده و خام باشند، به‌ویژه برای آزمایش شگفت‌آور مفیدند.

اما بزرگ‌ترین محدودیت ابزارهای Unix این است که فقط روی یک ماشین اجرا می‌شوند — و اینجاست که ابزارهایی مثل Hadoop وارد می‌شوند.

MapReduce و filesystemهای توزیع‌شده

MapReduce کمی شبیه ابزارهای Unix است، اما روی احتمالاً هزاران ماشین توزیع شده. مثل ابزارهای Unix، ابزار نسبتاً خام و brute-force اما شگفت‌آور مؤثر است. یک job MapReduce قابل مقایسه با یک process Unix است: یک یا چند ورودی می‌گیرد و یک یا چند خروجی تولید می‌کند.

مثل بیشتر ابزارهای Unix، اجرای job MapReduce معمولاً ورودی را تغییر نمی‌دهد و side effect جز تولید خروجی ندارد. فایل‌های خروجی یک‌بار و به‌صورت sequential نوشته می‌شوند (بخشی از فایل که نوشته شده بعداً تغییر نمی‌کند).

در حالی که ابزارهای Unix از stdin و stdout به‌عنوان ورودی و خروجی استفاده می‌کنند، jobهای MapReduce از فایل‌های filesystem توزیع‌شده می‌خوانند و می‌نویسند. در پیاده‌سازی Hadoop MapReduce، آن filesystem HDFS (Hadoop Distributed File System) است، reimplementation متن‌باز Google File System (GFS) [۱۹].

filesystemهای توزیع‌شده دیگری غیر از HDFS وجود دارد، مثل GlusterFS و Quantcast File System (QFS) [۲۰]. سرویس‌های object storage مثل Amazon S3، Azure Blob Storage و OpenStack Swift [۲۱] در بسیاری موارد مشابه‌اند.iv در این فصل بیشتر HDFS را مثال می‌گیریم، اما اصول برای هر filesystem توزیع‌شده اعمال می‌شود.

HDFS بر اصل shared-nothing استوار است (مقدمه بخش II)، برخلاف رویکرد shared-disk در NAS و SAN. ذخیره‌سازی shared-disk با appliance متمرکز، اغلب سخت‌افزار سفارشی و زیرساخت شبکه مثل Fibre Channel پیاده می‌شود. رویکرد shared-nothing سخت‌افزار خاص لازم ندارد، فقط رایانه‌های متصل با شبکه معمولی دیتاسنتر.

HDFS از process daemon روی هر ماشین تشکیل شده که سرویس شبکه‌ای ارائه می‌دهد تا گره‌های دیگر به فایل‌های روی آن ماشین دسترسی داشته باشند (فرض می‌کنیم هر ماشین general-purpose در دیتاسنتر دیسک دارد). سرور مرکزی به نام NameNode ردیابی می‌کند کدام block فایل روی کدام ماشین است. پس HDFS مفهوماً یک filesystem بزرگ می‌سازد که از فضای دیسک همه ماشین‌های daemon استفاده می‌کند.

برای تحمل خرابی ماشین و دیسک، blockهای فایل روی چند ماشین replicate می‌شوند. replication ممکن است چند کپی یکسان روی چند ماشین باشد (فصل ۵) یا erasure coding مثل Reed–Solomon که با overhead ذخیره‌سازی کمتر از replication کامل داده از دست رفته را بازیابی می‌کند [۲۰، ۲۲]. تکنیک‌ها شبیه RAID است که redundancy روی چند دیسک یک ماشین می‌دهد؛ تفاوت این است که در filesystem توزیع‌شده، دسترسی فایل و replication روی شبکه معمولی دیتاسنتر بدون سخت‌افزار خاص انجام می‌شود.

iv. یک تفاوت: با HDFS می‌توان task محاسباتی را روی ماشینی که کپی فایل را دارد زمان‌بندی کرد، در حالی که object storeها معمولاً storage و computation را جدا نگه می‌دارند. خواندن از دیسک محلی مزیت عملکرد دارد اگر پهنای باند شبکه گلوگاه باشد. اما با erasure coding مزیت locality از بین می‌رود چون داده از چند ماشین باید ترکیب شود تا فایل اصلی بازسازی شود [۲۰].

HDFS خوب scale کرده: در زمان نگارش، بزرگ‌ترین استقرارهای HDFS روی ده‌ها هزار ماشین با ظرفیت ترکیبی صدها petabyte اجرا می‌شوند [۲۳]. این مقیاس بزرگ به‌خاطر هزینه پایین‌تر ذخیره و دسترسی داده روی HDFS با سخت‌افزار commodity و نرم‌افزار متن‌باز نسبت به ظرفیت معادل روی appliance ذخیره‌سازی اختصاصی [۲۴] ممکن شده.

اجرای job MapReduce

MapReduce فریم‌ورک برنامه‌نویسی است که با آن روی filesystem توزیع‌شده مثل HDFS dataset بزرگ پردازش می‌کنید. ساده‌ترین راه فهم آن بازگشت به مثال تحلیل log وب‌سرور در «تحلیل ساده log» در صفحه ۳۹۱ است. الگوی پردازش داده در MapReduce بسیار شبیه آن مثال است:

1. مجموعه فایل ورودی خوانده و به رکورد شکسته می‌شود. در مثال log وب‌سرور، هر رکورد یک خط log است (یعنی \n جداکننده رکورد است).
2. تابع mapper برای هر رکورد ورودی فراخوانی می‌شود تا key و value استخراج کند. در مثال قبل، mapper همان awk '{print $7}' است: URL ($7) را key می‌گیرد و value خالی می‌ماند.
3. همه جفت‌های key-value بر اساس key مرتب می‌شوند. در مثال log، این کار با دستور sort اول انجام می‌شود.
4. تابع reducer روی جفت‌های key-value مرتب iterate می‌کند. اگر چند occurrence از همان key باشد، مرتب‌سازی آن‌ها را مجاور کرده و ترکیب بدون state زیاد در حافظه آسان است. در مثال، reducer با uniq -c پیاده شده که تعداد رکوردهای مجاور با همان key را می‌شمارد.

این چهار گام در یک job MapReduce انجام می‌شود. گام‌های ۲ (map) و ۴ (reduce) جایی است کد پردازش سفارشی می‌نویسید. گام ۱ (شکستن فایل به رکورد) توسط parser فرمت ورودی انجام می‌شود. گام ۳ (مرتب‌سازی) در MapReduce implicit است — نمی‌نویسید چون خروجی mapper همیشه قبل از reducer مرتب می‌شود.

برای ساخت job MapReduce دو callback پیاده می‌کنید: mapper و reducer (به «کوئری‌زدن MapReduce» در صفحه ۴۶ هم مراجعه کنید):

Mapper mapper برای هر رکورد ورودی یک‌بار فراخوانی می‌شود و key و value را از رکورد استخراج می‌کند. برای هر ورودی می‌تواند هر تعداد جفت key-value (از جمله صفر) تولید کند. state از یک رکورد به بعدی نگه نمی‌دارد، پس هر رکورد مستقل پردازش می‌شود.

Reducer فریم‌ورک MapReduce جفت‌های key-value تولید mapperها را جمع می‌کند، همه valueهای متعلق به یک key را کنار هم می‌گذارد و reducer را با iterator روی آن مجموعه فراخوانی می‌کند. reducer می‌تواند رکورد خروجی تولید کند (مثل تعداد occurrence یک URL).

در مثال log وب‌سرور، دستور sort دوم در گام ۵ URLها را بر اساس تعداد درخواست رتبه‌بندی می‌کرد. در MapReduce اگر مرحله مرتب‌سازی دوم لازم باشد، job MapReduce دوم می‌نویسید و خروجی job اول را ورودی آن می‌کنید. از این منظر، mapper داده را برای مرتب‌سازی آماده می‌کند و reducer داده مرتب‌شده را پردازش می‌کند.

اجرای توزیع‌شده MapReduce

تفاوت اصلی با pipeline دستورات Unix این است که MapReduce محاسبه را روی بسیاری ماشین موازی می‌کند بدون نوشتن کد صریح برای parallelism. mapper و reducer فقط یک رکورد در هر بار پردازش می‌کنند؛ نمی‌دانند ورودی از کجا و خروجی به کجا می‌رود، پس فریم‌ورک پیچیدگی جابه‌جایی داده بین ماشین‌ها را مدیریت می‌کند.

می‌توان ابزارهای استاندارد Unix را mapper و reducer در محاسبه توزیع‌شده استفاده کرد [۲۵]، اما معمولاً به‌صورت تابع در زبان برنامه‌نویسی conventional پیاده می‌شوند. در Hadoop MapReduce، mapper و reducer هر کدام کلاس Java با interface مشخص هستند. در MongoDB و CouchDB، mapper و reducer تابع JavaScript هستند (به «کوئری‌زدن MapReduce» در صفحه ۴۶).

شکل ۱۰-۱. یک job MapReduce با سه mapper و سه reducer.

موازی‌سازی بر partitioning استوار است (فصل ۶): ورودی job معمولاً directory در HDFS است و هر فایل یا block فایل در directory ورودی partition جداگانه‌ای است که map task جداگانه پردازش می‌کند (در شکل ۱۰-۱ با m 1، m 2 و m 3).

هر فایل ورودی معمولاً صدها مگابایت است. scheduler MapReduce (در نمودار نیست) سعی می‌کند هر mapper روی یکی از ماشین‌هایی که replica فایل ورودی را دارد اجرا شود، به‌شرط RAM و CPU کافی [۲۶]. این اصل «قرار دادن محاسبه نزدیک داده» [۲۷] است: از کپی فایل ورودی روی شبکه جلوگیری و بار شبکه کم و locality بیشتر می‌شود.

در بیشتر موارد، کد برنامه map task هنوز روی ماشین assign‌شده نیست، پس فریم‌ورک MapReduce ابتدا کد (مثلاً فایل JAR برای Java) را کپی می‌کند، map task را شروع می‌کند و فایل ورودی را می‌خواند و رکورد به رکورد به callback mapper می‌دهد. خروجی mapper جفت key-value است.

سمت reduce هم partition می‌شود. تعداد map task به block فایل ورودی بستگی دارد، اما تعداد reduce task توسط نویسنده job پیکربندی می‌شود (می‌تواند با map task متفاوت باشد). برای اینکه همه جفت key-value با یک key به یک reducer برسند، فریم‌ورک از hash کلید استفاده می‌کند تا reduce task مقصد را تعیین کند (به «پارتیشن‌بندی بر اساس hash کلید» در صفحه ۲۰۳).

جفت key-value باید مرتب شوند، اما dataset احتمالاً برای مرتب‌سازی معمول روی یک ماشین بزرگ است. مرتب‌سازی مرحله‌ای انجام می‌شود. ابتدا هر map task خروجی را بر اساس hash کلید به reducer partition می‌کند. هر partition در دیسک محلی mapper به فایل مرتب نوشته می‌شود، با تکنیک مشابه «SSTables و LSM-Trees» در صفحه ۷۶.

وقتی mapper خواندن فایل ورودی و نوشتن فایل خروجی مرتب را تمام کرد، scheduler MapReduce به reducerها می‌گوید می‌توانند فایل خروجی آن mapper را بگیرند. reducerها به هر mapper وصل می‌شوند و فایل جفت key-value مرتب partition خود را دانلود می‌کنند. partition بر اساس reducer، مرتب‌سازی و کپی partition از mapper به reducer shuffle [۲۶] نامیده می‌شود (نام گیج‌کننده — برخلاف shuffle deck کارت، randomness در MapReduce نیست).

reduce task فایل‌های mapper را merge می‌کند و ترتیب مرتب را حفظ می‌کند. پس اگر mapperهای مختلف رکورد با یک key تولید کرده باشند، در ورودی merge‌شده reducer مجاورند.

reducer با key و iterator که تدریجی همه رکوردهای همان key را scan می‌کند فراخوانی می‌شود (گاهی همه در حافظه جا نمی‌شوند). reducer با منطق دلخواه پردازش می‌کند و هر تعداد رکورد خروجی تولید می‌کند. این رکوردها در filesystem توزیع‌شده نوشته می‌شوند (معمولاً یک کپی روی دیسک محلی ماشین reducer و replica روی ماشین‌های دیگر).

workflowهای MapReduce

دامنه مسائلی که با یک job MapReduce حل می‌شوند محدود است. در مثال log، یک job MapReduce تعداد page view هر URL را می‌دهد، اما نه محبوب‌ترین URLها چون مرتب‌سازی دوم لازم است.

پس رایج است jobهای MapReduce در workflow زنجیره شوند و خروجی یک job ورودی job بعدی شود. فریم‌ورک Hadoop MapReduce پشتیبانی خاص workflow ندارد؛ زنجیره‌سازی implicit با نام directory است: job اول باید خروجی را در directory مشخص HDFS بنویسد و job دوم همان directory را ورودی بخواند. از دید فریم‌ورک، دو job مستقل‌اند.

workflowهای MapReduce زنجیره‌شده پس کمتر شبیه pipeline دستورات Unix (خروجی یک process مستقیم با buffer کوچک حافظه به process بعد) و بیشتر شبیه دنباله دستوراتی است که هر کدام خروجی را در فایل موقت می‌نویسد و دستور بعد از آن می‌خواند. مزیت و معایب دارد که در «materialization حالت میانی» در صفحه ۴۱۹ بحث می‌کنیم.

خروجی job دسته‌ای فقط وقتی معتبر است که job با موفقیت تمام شده باشد (MapReduce خروجی جزئی job شکست‌خورده را دور می‌ریزد). پس job در workflow فقط وقتی شروع می‌شود که jobهای قبلی — jobهایی که directory ورودی آن را تولید می‌کنند — با موفقیت تمام شده باشند. برای وابستگی‌های اجرا، schedulerهای workflow مختلف برای Hadoop ساخته شده: Oozie، Azkaban، Luigi، Airflow و Pinball [۲۸].

این schedulerها قابلیت مدیریت برای نگهداری مجموعه بزرگ job دسته‌ای هم دارند. workflow با ۵۰ تا ۱۰۰ job MapReduce برای ساخت سیستم توصیه [۲۹] رایج است و در سازمان بزرگ، تیم‌های مختلف jobهایی اجرا می‌کنند که خروجی یکدیگر را می‌خوانند. پشتیبانی ابزار برای مدیریت dataflow پیچیده مهم است.

ابزارهای سطح بالاتر Hadoop مثل Pig [۳۰]، Hive [۳۱]، Cascading [۳۲]، Crunch [۳۳] و FlumeJava [۳۴] workflow چندمرحله‌ای MapReduce می‌سازند که خودکار به‌درستی wired up می‌شوند.

joinها و grouping سمت reduce

joinها را در فصل ۲ در زمینه مدل داده و زبان کوئری دیدیم، اما پیاده‌سازی واقعی را بررسی نکردیم. وقت آن است دوباره ادامه دهیم.

در بسیاری datasetها رایج است یک رکورد با رکورد دیگر مرتبط باشد: foreign key در مدل relational، document reference در مدل document، یا edge در مدل graph. join لازم است هر وقت کدی باید به رکوردهای هر دو طرف association دسترسی داشته باشد (هم رکورد نگه‌دارنده reference و هم referenced). همان‌طور که در فصل ۲ دیدیم، denormalization نیاز join را کم می‌کند اما معمولاً کاملاً حذف نمی‌کند.v

در پایگاه داده، اگر کوئری فقط تعداد کمی رکورد دارد، پایگاه داده معمولاً با index سریع رکوردها را پیدا می‌کند (فصل ۳). اگر join دارد، ممکن است چند lookup index لازم باشد. اما MapReduce مفهوم index — حداقل به معنای معمول — ندارد.

وقتی job MapReduce مجموعه فایل ورودی می‌گیرد، کل محتوای همه فایل‌ها را می‌خواند؛ پایگاه داده این را full table scan می‌نامد. اگر فقط تعداد کمی رکورد می‌خواهید، full table scan نسبت به lookup index به‌شدت گران است. اما در کوئری analytic (به «پردازش تراکنش یا analytics؟» در صفحه ۹۰) رایج است aggregate روی تعداد زیاد رکورد محاسبه شود. در این حالت scan کل ورودی منطقی است، به‌ویژه اگر پردازش روی چند ماشین موازی شود.

v. joinهای این کتاب معمولاً equi-join هستند: رکورد با رکوردهایی که مقدار یکسان در فیلد خاص (مثل ID) دارند مرتبط می‌شود. بعضی پایگاه‌ها join عمومی‌تر (مثلاً با عملگر کمتر از به‌جای equality) پشتیبانی می‌کنند، اما فضای کتاب برای پوشش آن‌ها نیست.

وقتی در پردازش دسته‌ای از join صحبت می‌کنیم، منظور resolve کردن همه occurrence یک association در dataset است. مثلاً فرض می‌کنیم job داده همه کاربران را هم‌زمان پردازش می‌کند، نه lookup داده یک کاربر خاص (که با index خیلی کارآمدتر است).

مثال: تحلیل رویدادهای فعالیت کاربر

مثال typical join در job دسته‌ای در شکل ۱۰-۲ است. سمت چپ log رویدادهای کاربران login‌شده در وب‌سایت (activity events یا clickstream) و سمت راست پایگاه داده کاربران. می‌توان این را بخشی از star schema دید (به «ستاره و برفک: schema برای analytics» در صفحه ۹۳): log رویداد fact table و پایگاه کاربر یکی از dimensionهاست.

شکل ۱۰-۲. join بین log رویداد فعالیت کاربر و پایگاه داده پروفایل کاربر.

task analytic ممکن است فعالیت کاربر را با اطلاعات پروفایل correlate کند: مثلاً اگر پروفایل سن یا تاریخ تولد دارد، سیستم تعیین کند کدام صفحات برای کدام گروه سنی محبوب‌ترند. اما رویداد فعالیت فقط user ID دارد، نه کل پروفایل. embed کردن پروفایل در هر رویداد فعالیت احتمالاً waste است. پس رویدادهای فعالیت باید با پایگاه پروفایل join شوند.

ساده‌ترین پیاده‌سازی join روی رویدادهای فعالیت یکی‌یکی iterate می‌کند و برای هر user ID از پایگاه داده (روی سرور راه‌دور) query می‌زند. ممکن است، اما عملکرد بسیار ضعیف: throughput با round-trip time به سرور پایگاه داده محدود می‌شود، cache محلی به توزیع داده بستگی دارد و query موازی زیاد پایگاه را overwhelm می‌کند [۳۵].

برای throughput خوب در فرآیند دسته‌ای، محاسبه تا حد ممکن local به یک ماشین باشد. درخواست random-access شبکه برای هر رکورد خیلی کند است. علاوه بر این، query پایگاه راه‌دور job دسته‌ای را nondeterministic می‌کند چون داده پایگاه راه‌دور ممکن است تغییر کند.

پس رویکرد بهتر کپی پایگاه کاربر (مثلاً از backup با ETL — به «انبارداری داده» در صفحه ۹۱) و قرار دادن در همان filesystem توزیع‌شده log فعالیت است. پایگاه کاربر در یک مجموعه فایل HDFS و رکوردهای فعالیت در مجموعه دیگر؛ MapReduce همه رکوردهای مرتبط را در یک جا جمع و کارآمد پردازش می‌کند.

joinهای sort-merge

به یاد آورید mapper key و value از هر رکورد ورودی استخراج می‌کند. در شکل ۱۰-۲، key همان user ID است: یک مجموعه mapper روی رویدادهای فعالیت (user ID به‌عنوان key و رویداد به‌عنوان value) و مجموعه دیگر روی پایگاه کاربر (user ID به‌عنوان key و تاریخ تولد به‌عنوان value). در شکل ۱۰-۳ نشان داده شده.

شکل ۱۰-۳. sort-merge join سمت reduce روی user ID. اگر datasetهای ورودی به چند فایل partition شده باشند، هر کدام با چند mapper موازی پردازش می‌شوند.

وقتی فریم‌ورک MapReduce خروجی mapper را بر key partition و key-valueها را مرتب می‌کند، همه رویدادهای فعالیت و رکورد کاربر با همان user ID در ورودی reducer مجاور می‌شوند. job MapReduce حتی می‌تواند رکوردها را طوری مرتب کند که reducer همیشه ابتدا رکورد پایگاه کاربر و بعد رویدادهای فعالیت به ترتیب timestamp ببیند — secondary sort [۲۶].

reducer منطق join را آسان انجام می‌دهد: برای هر user ID یک‌بار فراخوانی می‌شود و با secondary sort، اولین value انتظار رکورد تاریخ تولد از پایگاه کاربر است. reducer تاریخ تولد را در متغیر محلی نگه می‌دارد و روی رویدادهای فعالیت با همان user ID iterate می‌کند و جفت viewed-url و viewer-age-in-years خروجی می‌دهد. jobهای MapReduce بعدی توزیع سن بیننده هر URL و cluster بر اساس گروه سنی را محاسبه می‌کنند.

چون reducer همه رکوردهای یک user ID یکجا پردازش می‌کند، فقط یک رکورد کاربر در حافظه لازم است و درخواست شبکه نمی‌زند. این الگوریتم sort-merge join است: خروجی mapper بر key مرتب و reducer لیست‌های مرتب هر دو طرف join را merge می‌کند.

جمع کردن داده مرتبط در یک جا

در sort-merge join، mapperها و مرتب‌سازی همه داده لازم join برای user ID مشخص را در یک جا — یک فراخوانی reducer — جمع می‌کنند. با آماده‌سازی داده، reducer می‌تواند کد single-threaded ساده با throughput بالا و overhead حافظه کم باشد.

یک نگاه به این معماری: mapperها «پیام» به reducer می‌فرستند. وقتی mapper جفت key-value emit می‌کند، key مثل آدرس مقصد value است. اگرچه key فقط رشته دلخواه است (نه آدرس شبکه مثل IP و port)، مثل آدرس رفتار می‌کند: همه جفت key-value با یک key به یک مقصد (فراخوانی reducer) می‌رسند.

مدل برنامه‌نویسی MapReduce جنبه فیزیکی ارتباط شبکه (رساندن داده به ماشین درست) را از منطق برنامه (پردازش داده وقتی دارید) جدا می‌کند. این با استفاده typical از پایگاه داده — درخواست fetch داده عمیق در کد برنامه [۳۶] — متفاوت است. چون MapReduce همه ارتباط شبکه را مدیریت می‌کند، کد برنامه از partial failure مثل crash گره دیگر محافظت می‌شود: MapReduce task شکست‌خورده را بدون تأثیر روی منطق برنامه retry می‌کند.

GROUP BY

علاوه بر join، کاربرد رایج الگوی «جمع کردن داده مرتبط در یک جا» grouping رکوردها بر اساس key (مثل بند GROUP BY در SQL) است. همه رکوردهای یک key یک گروه می‌سازند و گام بعد اغلب aggregation در هر گروه است — مثلاً:

• شمارش رکوردها در هر گروه (مثل شمارش page view که در SQL COUNT(*) است)
• جمع فیلد خاص (SUM(fieldname) در SQL)
• انتخاب k رکورد برتر بر اساس تابع رتبه‌بندی

ساده‌ترین پیاده‌سازی grouping با MapReduce: mapperها key-value با key grouping دلخواه تولید کنند. partition و مرتب‌سازی همه رکوردهای یک key را در یک reducer جمع می‌کند. پس grouping و join روی MapReduce شبیه به نظر می‌رسند.

کاربرد رایج دیگر grouping، جمع کردن همه رویدادهای فعالیت یک session کاربر برای یافتن توالی اقدامات — sessionization [۳۷] — است. مثلاً برای اینکه آیا کاربران نسخه جدید وب‌سایت بیشتر خرید می‌کنند (A/B testing) یا فعالیت بازاریابی ارزش دارد.

اگر چند وب‌سرور درخواست کاربر را handle کند، رویدادهای یک کاربر احتمالاً در log سرورهای مختلف پراکنده است. sessionization با session cookie، user ID یا شناسه مشابه به‌عنوان grouping key همه رویدادهای یک کاربر را در یک جا جمع و رویداد کاربران مختلف را در partition مختلف توزیع می‌کند.

مدیریت skew

الگوی «همه رکوردهای یک key در یک جا» وقتی داده زیادی به یک key مرتبط باشد می‌شکند. مثلاً در شبکه اجتماعی بیشتر کاربران چندصد ارتباط دارند، اما تعداد کمی celebrity میلیون‌ها follower دارند. چنین رکوردهای فعال linchpin object [۳۸] یا hot key نامیده می‌شوند.

جمع کردن همه فعالیت مربوط به celebrity (مثل reply به پست) در یک reducer به skew (یا hot spot) منجر می‌شود — reducerی که بسیار بیشتر از بقیه پردازش می‌کند (به «بار کاری skewed و تسکین hot spot» در صفحه ۲۰۵). چون job MapReduce وقتی تمام است که همه mapper و reducer تمام شده باشند، job بعدی منتظر کندترین reducer می‌ماند.

اگر ورودی join hot key دارد، الگوریتم‌هایی برای جبران وجود دارد. مثلاً skewed join در Pig ابتدا job sampling می‌زند تا hot keyها را پیدا کند [۳۹]. در join واقعی، mapper رکوردهای hot key را به یکی از چند reducer تصادفی می‌فرستد (برخلاف MapReduce معمول که reducer را deterministically از hash key انتخاب می‌کند). برای ورودی دیگر join، رکوردهای hot key باید به همه reducerهای آن key replicate شوند [۴۰].

این تکنیک کار hot key را روی چند reducer پخش و موازی‌سازی بهتر می‌کند، با هزینه replicate ورودی دیگر join به چند reducer. sharded join در Crunch مشابه است اما hot key باید صریح مشخص شود نه با sampling. شبیه «بار کاری skewed و تسکین hot spot» در صفحه ۲۰۵ با randomization برای hot spot در پایگاه partition‌شده.

skewed join optimization در Hive رویکرد دیگر دارد: hot key باید در metadata جدول صریح باشد و رکوردهای مرتبط در فایل جدا از بقیه ذخیره می‌شوند. در join روی آن جدول، برای hot key از map-side join (بخش بعد) استفاده می‌کند.

وقتی رکوردها بر hot key group و aggregate می‌شوند، grouping در دو مرحله انجام می‌شود: MapReduce اول رکوردها را به reducer تصادفی می‌فرستد تا هر reducer روی زیرمجموعه‌ای از hot key group کند و value aggregate فشرده‌تر per key خروجی دهد. MapReduce دوم valueهای reducerهای مرحله اول را به یک value per key ترکیب می‌کند.

joinهای سمت map

الگوریتم‌های join بخش قبل منطق join واقعی را در reducer انجام می‌دهند و reduce-side join نامیده می‌شوند. mapperها داده ورودی را آماده می‌کنند: key و value از هر رکورد، assign partition reducer و مرتب‌سازی بر key.

رویکرد reduce-side فرضی درباره داده ورودی نمی‌گیرد: هر property و ساختاری، mapper داده را برای join آماده می‌کند. اما معایب: مرتب‌سازی، کپی به reducer و merge ورودی reducer گران است. بسته به buffer حافظه، داده ممکن است چند بار روی دیسک نوشته شود [۳۷].

اگر درباره داده ورودی فرض بگیرید، join با map-side join سریع‌تر می‌شود: job MapReduce ساده‌شده بدون reducer و بدون مرتب‌سازی. هر mapper فقط یک block فایل ورودی از filesystem توزیع‌شده می‌خواند و یک فایل خروجی می‌نویسد — همین.

broadcast hash join

ساده‌ترین map-side join وقتی dataset بزرگ با dataset کوچک join می‌شود. dataset کوچک باید آن‌قدر کوچک باشد که در حافظه هر mapper جا شود.

در شکل ۱۰-۲، اگر پایگاه کاربر در حافظه جا شود، mapper هنگام شروع ابتدا پایگاه کاربر را از filesystem توزیع‌شده در hash table در حافظه می‌خواند. بعد رویدادهای فعالیت را scan می‌کند و user ID هر رویداد را در hash table lookup می‌کند.vi

هنوز چند map task می‌تواند باشد: یکی برای هر block فایل ورودی بزرگ join (در شکل ۱۰-۲، رویدادهای فعالیت ورودی بزرگ‌اند). هر mapper ورودی کوچک را کامل در حافظه بار می‌کند.

این الگوریتم ساده اما مؤثر broadcast hash join است: broadcast یعنی mapper هر partition ورودی بزرگ کل ورودی کوچک را می‌خواند (ورودی کوچک effectively به همه partition ورودی بزرگ broadcast می‌شود) و hash به hash table اشاره دارد. Pig («replicated join»)، Hive («MapJoin»)، Cascading و Crunch پشتیبانی می‌کنند. در موتورهای انبار داده مثل Impala [۴۱] هم استفاده می‌شود.

به‌جای hash table در حافظه، می‌توان ورودی کوچک join را در index فقط-خواندنی روی دیسک محلی ذخیره کرد [۴۲]. بخش‌های پرکاربرد index در page cache سیستم‌عامل می‌مانند و lookup تقریباً به‌سرعت hash table در حافظه بدون نیاز به جا شدن dataset در حافظه.

vi. این مثال فرض می‌کند برای هر key دقیقاً یک entry در hash table هست (با پایگاه کاربر درست است). عموماً hash table چند entry با یک key دارد و join operator همه matchها را خروجی می‌دهد.

partitioned hash join

اگر ورودی‌های map-side join یکسان partition شده باشند، hash join روی هر partition مستقل اعمال می‌شود. در شکل ۱۰-۲، رویدادهای فعالیت و پایگاه کاربر بر اساس رقم آخر user ID partition شوند (۱۰ partition هر طرف). mapper 3 ابتدا همه کاربران با ID پایان ۳ را در hash table بار می‌کند و رویدادهای کاربران با ID پایان ۳ را scan می‌کند.

اگر partitioning درست باشد، مطمئنید همه رکوردهایی که ممکن است join شوند در partition شماره یکسان‌اند و mapper فقط یک partition از هر dataset ورودی می‌خواند. هر mapper مقدار کمتری در hash table بار می‌کند.

این فقط وقتی کار می‌کند هر دو ورودی join تعداد partition یکسان، assign بر اساس همان key و همان hash function داشته باشند. اگر ورودی‌ها از jobهای MapReduce قبلی با همین grouping تولید شده باشند، فرض معقولی است. partitioned hash join در Hive bucketed map join [۳۷] نامیده می‌شود.

map-side merge join

variant دیگر map-side join وقتی datasetهای ورودی نه فقط یکسان partition شده‌اند بلکه بر همان key مرتب‌اند. دیگر مهم نیست ورودی در حافظه جا شود چون mapper همان mergeی را انجام می‌دهد که reducer معمولاً می‌کند: هر دو فایل ورودی را incrementally به ترتیب key صعودی می‌خواند و رکوردهای یک key را match می‌کند.

اگر map-side merge join ممکن باشد، احتمالاً jobهای MapReduce قبلی dataset را partitioned و sorted کرده‌اند. در اصل join در reduce stage job قبلی هم ممکن بود. اما merge join در job map-only جدا مناسب است اگر dataset partitioned و sorted برای کار دیگر هم لازم باشد.

workflowهای MapReduce با map-side join

وقتی خروجی join MapReduce توسط jobهای downstream مصرف می‌شود، انتخاب map-side یا reduce-side join ساختار خروجی را تعیین می‌کند. خروجی reduce-side join بر key partition و sort شده؛ خروجی map-side join مثل ورودی بزرگ partition و sort شده (یک map task برای هر block فایل ورودی بزرگ join، چه partitioned چه broadcast join).

map-side join فرض بیشتری درباره اندازه، مرتب‌سازی و partitioning dataset ورودی می‌گیرد. layout فیزیکی dataset در filesystem برای بهینه‌سازی join مهم است: کافی نیست فرمت encoding و نام directory را بدانید؛ تعداد partition و keyهای partition و sort را هم باید بدانید.

در اکوسystem Hadoop، metadata partitioning dataset اغلب در HCatalog و Hive metastore [۳۷] نگه‌داری می‌شود.

خروجی workflowهای دسته‌ای

زیاد درباره الگوریتم‌های workflow jobهای MapReduce صحبت کردیم، اما سؤال مهمی را نادیده گرفتیم: نتیجه همه آن پردازش چیست؟ چرا اصلاً این jobها را اجرا می‌کنیم؟

در کوئری پایگاه داده، پردازش تراکنش (OLTP) را از analytics جدا کردیم (به «پردازش تراکنش یا analytics؟» در صفحه ۹۰). کوئری OLTP معمولاً تعداد کمی رکورد را با key و index lookup می‌کند تا به کاربر نشان دهد (مثلاً در صفحه وب). کوئری analytic اغلب تعداد زیاد رکورد را scan می‌کند، grouping و aggregation انجام می‌دهد و خروجی اغلب گزارش است: نمودار تغییر metric در زمان، ۱۰ مورد برتر بر اساس رتبه، یا breakdown quantity به زیردسته. مصرف‌کننده گزارش اغلب analyst یا manager است.

پردازش دسته‌ای کجا می‌نشیند؟ نه پردازش تراکنش و نه analytics خالص. به analytics نزدیک‌تر است چون فرآیند دسته‌ای معمولاً بخش بزرگی از dataset ورودی را scan می‌کند. اما workflow jobهای MapReduce همان کوئری SQL analytic نیست (به «مقایسه Hadoop با پایگاه‌های داده توزیع‌شده» در صفحه ۴۱۴). خروجی فرآیند دسته‌ای اغلب گزارش نیست، بلکه ساختار دیگری است.

ساخت index جستجو

استفاده اصلی Google از MapReduce ساخت index موتور جستجو بود، به‌صورت workflow ۵ تا ۱۰ job MapReduce [۱]. Google بعداً از MapReduce برای این کار دور شد [۴۳]، اما MapReduce را از منظر ساخت index جستجو فهمیدن آسان‌تر است. (امروز Hadoop MapReduce هنوز راه خوبی برای index Lucene/Solr است [۴۴].)

در «جستجوی full-text و indexهای fuzzy» در صفحه ۸۸ خلاصه دیدیم index full-text مثل Lucene چگونه کار می‌کند: فایلی (term dictionary) که keyword را lookup می‌کنید و لیست document ID حاوی آن keyword (postings list) را می‌یابید. نمای ساده‌شده است — واقعاً داده اضافی برای رتبه relevance، تصحیح غلط املایی، synonym و غیره لازم است — اما اصل برقرار است.

اگر full-text search روی مجموعه ثابت document لازم است، فرآیند دسته‌ای راه مؤثری برای ساخت index است: mapperها document را partition می‌کنند، هر reducer index partition خود را می‌سازد و فایل index در filesystem توزیع‌شده نوشته می‌شود. index partitioned document (به «پارتیشن‌بندی و indexهای ثانویه» در صفحه ۲۰۶) خوب موازی می‌شود.

query index جستجو با keyword عملیات فقط-خواندنی است، پس فایل index پس از ساخت immutable است.

اگر مجموعه document indexed تغییر کند، یک گزینه دوره‌ای کل workflow indexing کل document را دوباره اجرا و فایل index قبلی را با جدید جایگزین کردن است. اگر فقط تعداد کمی document تغییر کرده گران است، اما فرآیند indexing ساده است: document ورودی، index خروجی.

گزینه دیگر index incrementally ساختن است. همان‌طور که در فصل ۳ دیدیم، برای add/remove/update document در index، Lucene segment جدید می‌نویسد و segmentها را در پس‌زمینه merge و compact می‌کند. پردازش incremental بیشتر در فصل ۱۱.

key-value store به‌عنوان خروجی فرآیند دسته‌ای

index جستجو فقط یک نمونه خروجی workflow پردازش دسته‌ای است. کاربرد رایج دیگر ساخت سیستم یادگیری ماشین مثل classifier (spam filter، anomaly detection، تشخیص تصویر) و سیستم توصیه (people you may know، محصولات مورد علاقه، جستجوهای مرتبط [۲۹]).

خروجی آن jobهای دسته‌ای اغلب نوعی پایگاه داده است: مثلاً پایگاهی که با user ID query شود و دوستان پیشنهادی بدهد، یا با product ID لیست محصولات مرتبط [۴۵].

این پایگاه‌ها باید از برنامه وب که درخواست کاربر را handle می‌کند query شوند که معمولاً از زیرساخت Hadoop جدا است. پس خروجی فرآیند دسته‌ای چگونه به پایگاهی برمی‌گردد که برنامه وب query کند؟

انتخاب واضح استفاده مستقیم client library پایگاه دلخواه در mapper یا reducer و نوشتن از job دسته‌ای به سرور پایگاه داده، رکورد به رکورد. کار می‌کند (اگر firewall از Hadoop به پایگاه production اجازه دهد)، اما برای چند دلیل ایده بد است:

• همان‌طور که در join گفتیم، درخواست شبکه برای هر رکورد چند مرتبه کندتر از throughput معمول task دسته‌ای است. حتی با batching در client library، عملکرد احتمالاً ضعیف است.
• job MapReduce اغلب taskهای زیاد موازی اجرا می‌کند. اگر همه mapper یا reducer هم‌زمان به یک پایگاه خروجی بنویسند با نرخ batch، پایگاه overwhelm می‌شود و عملکرد query سایر بخش سیستم [۳۵] آسیب می‌بیند.
• MapReduce معمولاً guarantee همه‌یا-هیچ برای خروجی job می‌دهد: اگر job موفق باشد، نتیجه اجرای دقیقاً یک‌بار هر task است حتی اگر بعضی task شکست خورده و retry شده باشند؛ اگر کل job شکست بخورد، خروجی نیست. اما نوشتن به سیستم خارجی از داخل job side effect خارجی visible تولید می‌کند که پنهان نمی‌شود. پس باید نگران نتیجه job نیمه‌تمام visible برای سیستم‌های دیگر و پیچیدگی task attempt و speculative execution Hadoop باشید.

راه‌حل بهتر ساخت پایگاه داده کاملاً جدید داخل job دسته‌ای و نوشتن به‌صورت فایل در directory خروجی job در filesystem توزیع‌شده، مثل index جستجو. آن فایل‌های داده پس از نوشتن immutable‌اند و bulk در سرورهای فقط-خواندنی load می‌شوند. key-value storeهای مختلف ساخت فایل پایگاه داده در job MapReduce را پشتیبانی می‌کنند از جمله Voldemort [۴۶]، Terrapin [۴۷]، ElephantDB [۴۸] و HBase bulk loading [۴۹].

ساخت این فایل پایگاه داده استفاده خوب MapReduce است: mapper key استخراج و sort بر key بخش زیادی از کار index است. چون بیشتر این key-value storeها فقط-خواندنی‌اند (فایل فقط یک‌بار توسط job دسته‌ای نوشته و immutable می‌شود)، ساختار داده ساده است. مثلاً WAL لازم نیست (به «قابل اعتماد کردن B-tree» در صفحه ۸۲).

هنگام load داده در Voldemort، سرور همچنان به فایل داده قدیمی سرویس می‌دهد در حالی که فایل جدید از filesystem توزیع‌شده به دیسک محلی کپی می‌شود. پس از کپی، سرور atomically به فایل جدید switch می‌کند. اگر مشکلی پیش آید، به فایل قدیمی immutable برمی‌گردد [۴۶].

فلسفه خروجی فرآیند دسته‌ای

فلسفه Unix که در «فلسفه Unix» در صفحه ۳۹۴ دیدیم با explicit بودن dataflow آزمایش را تشویق می‌کند: برنامه ورودی می‌خواند و خروجی می‌نویسد. ورودی دست‌نخورده می‌ماند، خروجی قبلی کاملاً با خروجی جدید جایگزین می‌شود و side effect دیگری نیست. یعنی می‌توانید دستور را هرچند بار rerun کنید بدون خراب کردن state سیستم.

مدیریت خروجی job MapReduce همان فلسفه را دنبال می‌کند. با immutable در نظر گرفتن ورودی و پرهیز از side effect (مثل نوشتن به پایگاه خارجی)، job دسته‌ای نه فقط عملکرد بهتر بلکه نگهداری آسان‌تر:

• اگر bug در کد باشد و خروجی اشتباه یا corrupt شود، به نسخه قبلی کد rollback و job rerun کنید و خروجی درست می‌شود. یا ساده‌تر، خروجی قدیمی را در directory دیگر نگه دارید و switch back کنید. پایگاه با تراکنش read-write این property را ندارد: اگر کد buggy داده بد بنویسد، rollback کد داده پایگاه را درست نمی‌کند. (ایده recover از کد buggy human fault tolerance [۵۰] نامیده شده.)
• به‌خاطر rollback آسان، توسعه feature سریع‌تر از محیطی که اشتباه irreversible damage می‌دهد. اصل کمینه کردن irreversibility برای Agile [۵۱] مفید است.
• اگر map یا reduce task شکست بخورد، فریم‌ورک MapReduce دوباره زمان‌بندی و روی همان ورودی اجرا می‌کند. اگر شکست از bug کد باشد، crash می‌کند و job بعد از چند تلاش fail می‌شود؛ اگر مشکل گذرا باشد، fault tolerate می‌شود. retry خودکار فقط چون ورودی immutable و خروجی task شکست‌خورده توسط فریم‌ورک discard می‌شود safe است.
• همان مجموعه فایل ورودی jobهای مختلف از جمله monitoring که metric محاسبه و خروجی job را با run قبلی مقایسه می‌کند.
• مثل ابزارهای Unix، job MapReduce منطق را از wiring (پیکربندی directory ورودی/خروجی) جدا می‌کند: separation of concerns و reuse کد — یک تیم jobی که یک کار را خوب انجام می‌دهد، تیم دیگر تصمیم می‌گیرد کجا و کی اجرا شود.

در این زمینه، اصول طراحی Unix برای Hadoop هم خوب کار می‌کند — اما Unix و Hadoop در بعضی چیزها متفاوت‌اند. مثلاً چون بیشتر ابزارهای Unix فایل متن untyped فرض می‌کنند، parse ورودی زیاد لازم است (مثال log از {print $7}). روی Hadoop بعضی تبدیل‌های syntactic سطح پایین با فرمت ساختاریافته‌تر حذف می‌شوند: Avro (به «Avro» در صفحه ۱۲۲) و Parquet (به «ذخیره‌سازی ستونی» در صفحه ۹۵) با encoding schema-based کارآمد و evolution schema (فصل ۴).

مقایسه Hadoop با پایگاه‌های داده توزیع‌شده

Hadoop کمی شبیه Unix توزیع‌شده است: HDFS filesystem و MapReduce پیاده‌سازی عجیب process Unix (که همیشه بین map و reduce utility sort اجرا می‌کند). join و grouping مختلف روی این primitiveها پیاده شد.

وقتی مقاله MapReduce [۱] منتشر شد، به‌نوعی جدید نبود. همه الگوریتم‌های پردازش و join موازی بخش‌های قبل دهه‌ها قبل در پایگاه‌های massively parallel processing (MPP) پیاده شده بودند [۳، ۴۰]. مثلاً Gamma، Teradata و Tandem NonStop SQL پیشگام بودند [۵۲].

بزرگ‌ترین تفاوت: MPP database روی اجرای موازی کوئری SQL analytic روی cluster تمرکز دارد، در حالی که MapReduce به‌علاوه filesystem توزیع‌شده [۱۹] چیزی شبیه سیستم‌عامل general-purpose برای اجرای برنامه دلخواه است.

تنوع storage

پایگاه داده داده را بر اساس مدل خاص (relational یا document) ساختار می‌دهد، اما فایل در filesystem توزیع‌شده فقط دنباله بایت است که با هر مدل و encoding نوشته می‌شود. ممکن است رکورد پایگاه داده باشد، یا متن، تصویر، ویدیو، reading سنسور، ماتریس sparse، feature vector، توالی genome یا هر داده دیگر.

به‌طور خلاصه، Hadoop امکان dump بی‌تفاوت داده در HDFS و بعد فهمیدن نحوه پردازش [۵۳] را باز کرد. MPP database معمولاً modeling دقیق upfront و import به فرمت proprietary لازم دارد.

از دید purist، modeling و import دقیق مطلوب است چون کاربران داده با کیفیت‌تر دارند. اما در عمل، در دسترس بودن سریع داده — حتی raw و سخت‌مصرف — اغلب از تصمیم مدل ایده‌آل upfront [۵۴] ارزشمندتر است.

ایده شبیه انبار داده (به «انبارداری داده» در صفحه ۹۱): جمع داده از بخش‌های سازمان در یک جا برای join datasetهای قبلاً جدا valuable است. schema دقیق MPP جمع‌آوری متمرکز را کند می‌کند؛ جمع raw و schema بعداً (data lake یا enterprise data hub [۵۵]) سریع‌تر است.

dump بی‌تفاوت بار تفسیر را به producer منتقل نمی‌کند: consumer باید تفسیر کند (schema-on-read [۵۶]؛ به «انعطاف schema در مدل document» در صفحه ۳۹). اگر producer و consumer تیم‌های مختلف با اولویت متفاوت باشند مزیت است. شاید یک مدل ایده‌آل نباشد بلکه viewهای مختلف برای اهداف مختلف. dump raw چند transformation را ممکن می‌کند. sushi principle: «raw data is better» [۵۷].

پس Hadoop اغلب برای ETL (به «انبارداری داده» در صفحه ۹۱): dump داده از OLTP به filesystem raw، job MapReduce برای پاک‌سازی، transform به فرم relational و import به MPP warehouse. modeling هنوز هست اما جدا از جمع‌آوری. decoupling چون filesystem هر encoding را پشتیبانی می‌کند.

تنوع مدل‌های پردازش

MPP database monolithic و tightly integrated است: layout storage، query planning، scheduling و execution. همه component برای نیاز database tune می‌شوند و عملکرد خوب روی کوئری طراحی‌شده. SQL expressive بدون کد و برای analyst با ابزار گرافیکی (Tableau) accessible است.

اما همه پردازش را نمی‌توان SQL بیان کرد. machine learning، recommendation، full-text search با relevance ranking، تحلیل تصویر — مدل پردازش general‌تر لازم است. این پردازش‌ها application-specific‌اند (feature engineering، مدل NLP، تخمین ریسک fraud) و کد لازم است نه فقط query.

MapReduce به engineer اجازه داد کد خود را روی dataset بزرگ اجرا کند. با HDFS و MapReduce می‌توان SQL engine ساخت — Hive [۳۱] این کار را کرد. اما batch processهای دیگری هم هست که SQL نیست.

بعداً MapReduce برای بعضی پردازش محدودکننده و کند بود؛ مدل‌های دیگر روی Hadoop (به «فراتر از MapReduce» در صفحه ۴۱۹). دو مدل SQL و MapReduce کافی نبود. openness Hadoop پیاده‌سازی range رویکردها را ممکن کرد که در MPP monolithic نبود [۵۸].

مدل‌های مختلف روی یک cluster shared-use، همه به همان فایل filesystem دسترسی. نیازی به import به سیستم‌های تخصصی مختلف نیست: flexible برای workload متنوع. کمتر جابه‌جایی داده، derive value و experiment با مدل جدید آسان‌تر.

اکوسystem Hadoop پایگاه OLTP random-access مثل HBase (به «SSTables و LSM-Trees» در صفحه ۷۶) و analytic MPP مثل Impala [۴۱] دارد. هیچ‌کدام MapReduce نمی‌زنند اما HDFS برای storage. رویکرد دسترسی و پردازش متفاوت اما coexist و integrate می‌شوند.

طراحی برای fault مکرر

دو تفاوت دیگر MapReduce و MPP: مدیریت fault و استفاده حافظه/دیسک. فرآیند دسته‌ای نسبت به سیستم آنلاین به fault حساس‌تر نیست چون fail فوری کاربر را تحت تأثیر نمی‌گذارد و همیشه دوباره اجرا می‌شود.

اگر گره در اجرای query crash کند، بیشتر MPP database کل query را abort و کاربر resubmit یا auto rerun می‌کند [۳]. query معمولاً چند ثانیه یا دقیقه است، پس retry قابل قبول است. MPP data را در حافظه نگه می‌دارد (hash join) تا از دیسک نخواند.

MapReduce شکست map یا reduce task را بدون تأثیر روی کل job با retry در granularity task tolerate می‌کند. eager به دیسک می‌نویسد، partly برای fault tolerance و partly چون dataset بزرگ‌تر از حافظه است.

رویکرد MapReduce برای job بزرگ‌تر مناسب‌تر است: داده زیاد و مدت طولانی که احتمالاً حداقل یک task fail شود. rerun کل job برای یک task fail waste است. recovery در granularity task با overhead برای پردازش بدون fault کندتر می‌شود، اما اگر نرخ fail task بالا باشد trade-off معقول است.

این فرض‌ها چقدر realistic‌اند؟ در بیشتر cluster خرابی ماشین هست اما نادر — شاید rare enough که بیشتر job fail task نبینند. آیا overhead fault tolerance ارزش دارد؟

برای sparing use حافظه و recovery سطح task MapReduce، محیط طراحی اصلی را ببینیم. Google datacenter mixed-use دارد: سرویس production آنلاین و job batch آفلاین روی همان ماشین. هر task resource allocation (CPU، RAM، disk) با container enforce می‌شود. priority دارد؛ task با priority بالاتر resource می‌خواهد، task با priority پایین‌تر preempt می‌شود. priority قیمت resource را تعیین می‌کند [۵۹].

معماری اجازه overcommit resource non-production (low-priority) می‌دهد چون سیستم می‌داند resource را reclaim می‌کند. overcommit utilization بهتر نسبت به segregate production/non-production. اما job MapReduce low-priority است و risk preempt هر لحظه. job batch «آشغال زیر میز» می‌خورد — resource باقی‌مانده بعد از high-priority.

در Google، task MapReduce یک ساعته تقریباً ۵٪ risk terminate برای priority بالاتر دارد — بیش از یک مرتبه بالاتر از fail سخت‌افزار یا reboot [۵۹]. با این نرخ preempt، job با ۱۰۰ task ۱۰ دقیقه‌ای، risk بیش از ۵۰٪ است حداقل یک task قبل از پایان terminate شود.

پس MapReduce برای terminate غیرمنتظره مکرر task طراحی شده — نه چون سخت‌افزار unreliable، بلکه چون آزادی terminate process utilization بهتر cluster را ممکن می‌کند.

در scheduler متن‌باز، preempt کمتر رایج است. CapacityScheduler YARN preempt برای balance queue [۵۸] دارد، اما priority preempt عمومی در YARN، Mesos یا Kubernetes در زمان نگارش نیست [۶۰]. وقتی task کمتر terminate می‌شود، تصمیمات طراحی MapReduce کمتر منطقی است. بخش بعد alternative به MapReduce.

فراتر از MapReduce

MapReduce در اواخر ۲۰۰۰ محبوب و hype زیاد داشت، اما فقط یکی از مدل‌های برنامه‌نویسی distributed است. بسته به حجم، ساختار داده و نوع پردازش، ابزار دیگر مناسب‌تر است.

با این حال زیاد درباره MapReduce صحبت کردیم چون ابزار یادگیری مفید است — abstraction نسبتاً روشن روی filesystem توزیع‌شده. simple در فهم نه در استفاده: پیاده‌سازی job پیچیده با API خام MapReduce سخت و پرکار است — مثلاً join algorithm از صفر [۳۷].

در پاسخ، مدل‌های سطح بالاتر (Pig، Hive، Cascading، Crunch) abstraction روی MapReduce ساخته شد. با فهم MapReduce، یادگیری آسان و construct سطح بالاتر task batch رایج را ساده‌تر می‌کند.

اما مشکل مدل اجرای MapReduce خودش است که abstraction دیگر حل نمی‌کند و برای بعضی پردازش عملکرد ضعیف — MapReduce robust است: dataset تقریباً arbitrary روی multi-tenant unreliable با terminate مکرر task تمام می‌شود (هرچند کند). ابزار دیگر گاه orders of magnitude سریع‌تر.

در ادامه فصل alternative batch processing. فصل ۱۱ stream processing — راه دیگر سرعت batch.

materialization حالت میانی

همان‌طور که گفتیم، هر job MapReduce از job دیگر مستقل است. تماس با دنیای بیرون: directory ورودی/خروجی filesystem توزیع‌شده. اگر خروجی job اول ورودی job دوم شود، directory ورودی job دوم = directory خروجی job اول و scheduler workflow job دوم را فقط بعد از تمام job اول start می‌کند.

اگر خروجی job اول dataset است که در سازمان publish می‌کنید، setup منطقی است: با نام reference و reuse برای jobهای مختلف (تیم‌های دیگر). publish در مکان شناخته‌شده filesystem loose coupling — job نمی‌داند چه کسی input تولید یا output مصرف می‌کند (به «جداسازی منطق و wiring» در صفحه ۳۹۶).

اما اغلب خروجی job فقط ورودی یک job دیگر همان تیم است. فایل filesystem فقط حالت میانی است: عبور داده از job به job بعد. در workflow پیچیده recommendation با ۵۰–۱۰۰ job MapReduce [۲۹]، حالت میانی زیاد است.

نوشتن حالت میانی به فایل materialization نامیده می‌شود (در view materialized در «تجمیع: Data Cube و viewهای materialized» در صفحه ۱۰۱: eager compute و نوشتن به‌جای compute on demand).

در مقابل، مثال log ابتدای فصل pipe Unix خروجی یک command را به ورودی بعدی وصل می‌کند. pipe حالت میانی را fully materialize نمی‌کند بلکه incrementally stream می‌کند با buffer کوچک حافظه.

materialization کامل حالت میانی MapReduce نسبت به pipe Unix معایب دارد:

• job MapReduce فقط وقتی start می‌شود همه task jobهای قبلی (که input تولید می‌کنند) تمام شده باشند؛ processهای pipe Unix هم‌زمان start و output به‌محض تولید consume می‌شود. skew یا load متفاوت روی ماشین‌ها یعنی straggler task که دیر تمام می‌شود؛ انتظار برای همه task job قبلی کل workflow را کند می‌کند.
• mapper اغلب redundant: همان فایلی که reducer نوشت را می‌خواند و برای partition/sort بعدی آماده می‌کند. mapper می‌تواند بخش reducer قبلی باشد: اگر خروجی reducer مثل خروجی mapper partition و sort شده، reducerها مستقیم chain شوند بدون mapper بینابین.
• حالت میانی در filesystem replicate روی چند گره است — برای داده موقت اغلب overkill.

موتورهای dataflow

برای رفع مشکل MapReduce، engine اجرای batch توزیع‌شده جدید ساخته شد: Spark [۶۱، ۶۲]، Tez [۶۳، ۶۴] و Flink [۶۵، ۶۶]. تفاوت طراحی دارند اما مشترک: کل workflow یک job است نه subjob مستقل.

چون جریان داده از چند stage را explicit مدل می‌کنند، موتور dataflow نامیده می‌شوند. مثل MapReduce با فراخوانی مکرر تابع کاربر برای یک رکورد در یک thread. موازی‌سازی با partition ورودی و کپی خروجی یک function روی شبکه به ورودی function دیگر.

برخلاف MapReduce، functionها نقش strict map/reduce متناوب لازم ندارند و flexible‌تر assemble می‌شوند — operator نامیده می‌شوند. موتور dataflow گزینه‌های اتصال خروجی operator به ورودی دیگر:

• repartition و sort رکورد بر key، مثل shuffle MapReduce (به «اجرای توزیع‌شده MapReduce» در صفحه ۴۰۰): sort-merge join و grouping مثل MapReduce.
• چند ورودی با partition یکسان بدون sort: partitioned hash join — partition مهم، order نه چون hash table order را randomize می‌کند.
• broadcast hash join: یک خروجی operator به همه partition operator join فرستاده می‌شود.

بر پایه Dryad [۶۷] و Nephele [۶۸]. مزیت نسبت به MapReduce:

• کار گران مثل sort فقط جایی که لازم است، نه همیشه بین map و reduce.
• map task غیرضروری نیست — کار mapper اغلب در reduce operator قبلی (mapper partitioning dataset را عوض نمی‌کند).
• join و dependency explicit — scheduler overview دارد و locality optimize می‌کند: task مصرف‌کننده داده روی همان ماشین producer.
• حالت میانی بین operator در حافظه یا دیسک محلی — I/O کمتر از HDFS (replicate و disk هر replica). MapReduce این را برای خروجی mapper دارد؛ dataflow engine generalize می‌کند.
• operator به‌محض آماده ورودی start — انتظار برای تمام stage قبلی نیست.
• reuse process JVM موجود — startup کمتر از MapReduce (JVM جدید هر task).

همان محاسبات workflow MapReduce با engine dataflow با optimizationهای بالا سریع‌تر. operator generalization map/reduce — کد Pig/Hive/Cascading با تغییر config از MapReduce به Tez/Spark بدون تغییر کد [۶۴].

Tez library نازک با YARN shuffle برای copy بین گره [۵۸]؛ Spark و Flink framework بزرگ با لایه شبکه، scheduler و API خود.

API سطح بالا را بعداً بحث می‌کنیم.

تحمل fault

مزیت materialization کامل حالت میانی به filesystem: durable و fault tolerance MapReduce آسان — task fail روی ماشین دیگر restart و همان input از filesystem.

Spark، Flink و Tez به HDFS نمی‌نویسند — fault متفاوت: ماشین fail و حالت میانی از دست رفت، recompute از داده موجود (stage میانی قبلی یا ورودی اصلی معمولاً HDFS).

برای recompute، فریم‌ورک lineage داده را track می‌کند — partition ورودی و operator اعمال‌شده. Spark با RDD [۶۱]؛ Flink checkpoint state operator [۶۶].

برای recompute مهم است computation deterministic باشد: همان ورودی، همان خروجی؟ اگر داده از دست‌رفته قبلاً به downstream فرستاده شده و recompute متفاوت باشد، downstream contradiction حل سخت است. برای operator nondeterministic معمولاً downstream هم kill و rerun.

پس operator deterministic بهتر — nondeterminism آسان creep می‌کند: iterate hash table بدون order تضمین‌شده، الگوریتم probabilistic با random، system clock یا منبع خارجی nondeterministic. با seed ثابت pseudorandom و حذف علل nondeterminism recover reliable.

recompute همیشه جواب نیست: اگر حالت میانی خیلی کوچک‌تر از source یا CPU-intensive، materialization به فایل ارزان‌تر از recompute.

بحث materialization

بازگشت به analogy Unix: MapReduce مثل نوشتن خروجی هر command به فایل موقت؛ dataflow engine شبیه pipe Unix. Flink حول pipelined execution: incrementally خروجی operator به operator دیگر بدون انتظار کامل ورودی.

sort باید کل ورودی را بخورد قبل از خروجی — آخرین رکورد ممکن است کمترین key باشد. operator نیازمند sort state موقت accumulate می‌کند. بقیه workflow pipelined.

وقتی job تمام شد، خروجی باید durable باشد — معمولاً دوباره filesystem توزیع‌شده. با dataflow engine، dataset materialized روی HDFS معمولاً ورودی و خروجی نهایی job. مثل MapReduce ورودی immutable و خروجی کاملاً جایگزین. بهبود: نوشتن همه حالت میانی به filesystem نیست.

graphها و پردازش iterative

در «مدل‌های داده شبیه graph» در صفحه ۴۹ graph برای modeling و زبان graph query برای traverse edge و vertex دیدیم — OLTP-style: کوئری سریع تعداد کم vertex.

graph در batch processing: پردازش offline یا تحلیل کل graph. در machine learning (recommendation، ranking). معروف‌ترین PageRank [۶۹]: محبوبیت صفحه وب از لینک صفحات دیگر — بخش formula رتبه‌بندی موتور جستجو.

موتور dataflow مثل Spark، Flink و Tez (به «materialization حالت میانی» در صفحه ۴۱۹) operator job را DAG می‌چینند. این graph processing نیست: در dataflow engine جریان داده بین operator graph است اما داده خود tuple relational-style است. در graph processing، داده خود graph است. naming confusion دیگر!

بسیاری الگوریتم graph با traverse یک edge در هر بار، join vertex با vertex مجاور برای propagate اطلاعات و تکرار تا شرط — مثلاً تا edge باقی نماند یا metric converge. شکل ۲-۶: transitive closure مکان‌های North America.

graph در filesystem توزیع‌شده (فایل vertex و edge) ذخیره می‌شود، اما «تکرار تا done» در MapReduce plain نیست — فقط یک pass. الگوریتم iterative:

1. scheduler خارجی batch process یک گام algorithm.
2. پس از تمام، scheduler بررسی finished (شرط completion).
3. اگر نه، برگشت به ۱.

کار می‌کند اما MapReduce برای iterative ناکارآمد — هر بار کل dataset ورودی می‌خواند و خروجی کاملاً جدید حتی اگر فقط بخش کوچک graph تغییر کرده.

مدل پردازش Pregel

بهینه‌سازی batch graph: مدل bulk synchronous parallel (BSP) [۷۰]. Apache Giraph [۳۷]، GraphX Spark و Gelly Flink [۷۱]. مدل Pregel از مقاله Pregel Google [۷۲].

در MapReduce mapper «پیام» به reducer می‌فرستد چون framework خروجی mapper با یک key جمع می‌کند. Pregel: vertex «پیام» به vertex دیگر، معمولاً along edge graph.

هر iteration تابع برای هر vertex با همه پیام‌های رسیده — شبیه reducer. تفاوت: vertex state را در حافظه iteration به iteration نگه می‌دارد؛ فقط پیام جدید پردازش. اگر پیامی در بخشی از graph نیست، کار نیست.

شبیه actor model (به «فریم‌ورک actor توزیع‌شده» در صفحه ۱۳۸) — هر vertex actor — اما state vertex و پیام fault-tolerant و durable؛ communication در round ثابت: هر iteration همه پیام iteration قبل deliver.

تحمل fault

vertex فقط با message passing (نه query مستقیم) — پیام batch و کمتر انتظار communication. تنها انتظار بین iteration: Pregel تضمین همه پیام یک iteration در iteration بعد؛ iteration قبل باید تمام و پیام روی شبکه copy شود.

شبکه ممکن است drop، duplicate یا delay (به «شبکه‌های غیرقابل اعتماد» در صفحه ۲۷۷)، اما Pregel پیام را exactly once در vertex مقصد iteration بعد پردازش می‌کند. مثل MapReduce، framework از fault recover می‌کند.

fault tolerance با checkpoint دوره‌ای state همه vertex پایان iteration — state کامل به storage durable. گره fail و state حافظه از دست رفت: rollback کل computation به checkpoint و restart؛ یا اگر deterministic و پیام log، فقط partition از دست‌رفته recover (مثل dataflow engine) [۷۲].

اجرای موازی

vertex نمی‌داند روی کدام ماشین فیزیکی است؛ پیام به vertex ID می‌فرستد. فریم‌ورک graph partition — vertex کدام ماشین، پیام چگونه route.

«thinking like a vertex» — partition arbitrary. ایده‌آل colocate vertexهای پرارتباط؛ پیدا کردن partition بهینه سخت — اغلب partition با vertex ID تصادفی بدون گروه‌بندی مرتبط.

الگوریتم graph overhead ارتباط cross-machine زیاد — حالت میانی (پیام بین گره) اغلب بزرگ‌تر از graph اصلی. شبکه distributed graph را کند می‌کند.

اگر graph در حافظه یک رایانه جا شود، الگوریتم single-machine (حتی single-thread) احتمالاً از batch توزیع‌شده سریع‌تر [۷۳، ۷۴]. اگر بزرگ‌تر از حافظه اما روی دیسک یک رایانه، GraphChi [۷۵]. اگر بزرگ‌تر از یک ماشین، Pregel لازم؛ parallelize graph ongoing research [۷۶].

API و زبان سطح بالا

از محبوبیت MapReduce، engine batch توزیع‌شده mature شد. زیرساخت robust برای petabyte روی ۱۰٬۰۰۰+ ماشین — مشکل فیزیکی batch «حل‌شده»؛ توجه به مدل برنامه‌نویسی، کارایی و گسترش مسائل.

Pig، Hive، Cascading، Crunch محبوب چون MapReduce دستی laborious است. با Tez، migration به dataflow بدون rewrite [۶۴]. Spark و Flink API dataflow سطح بالا با الهام FlumeJava [۳۴].

API dataflow building block relational-style: join dataset بر فیلد؛ group tuple بر key؛ filter؛ aggregate (count، sum، …). داخلاً join و grouping algorithms این فصل.

مزیت: کد کمتر؛ استفاده interactive — کد incremental در shell و observe. شبیه فلسفه Unix («فلسفه Unix» صفحه ۳۹۴).

API سطح بالا human productive و کارایی machine: optimizer job.

حرکت به زبان کوئری declarative

join به‌عنوان operator relational — فریم‌ورک property ورودی را analyze و بهترین join algorithm را انتخاب می‌کند. Hive، Spark و Flink cost-based optimizer [۶۶، ۷۷، ۷۸، ۷۹] و ترتیب join برای minimize حالت میانی.

انتخاب join algorithm تفاوت بزرگ عملکرد — nice که همه algorithm را حفظ نکنید. declarative: application join لازم را state می‌کند، optimizer اجرا — «زبان‌های کوئری برای داده» صفحه ۴۲.

MapReduce و dataflow با SQL fully declarative متفاوت: MapReduce callback function — mapper/reducer برای هر رکورد/گروه؛ function کد arbitrary. مزیت: ecosystem کتابخانه (parse، NLP، تصویر، آماری).

آزادی کد arbitrary MapReduce را از MPP جدا کرد («مقایسه Hadoop با پایگاه‌های داده توزیع‌شده» صفحه ۴۱۴) — UDF در database cumbersome و integrate ضعیف با Maven/npm/Rubygems.

dataflow engine declarative بیشتر — filter/map ساده در callback overhead CPU هر رکورد. declarative: optimizer column-oriented storage («ذخیره‌سازی ستونی» صفحه ۹۵) و فقط ستون لازم. Hive، Spark DataFrames، Impala vectorized execution («پهنای باند حافظه و پردازش vectorized» صفحه ۹۹). Spark bytecode JVM [۷۹]؛ Impala LLVM native code [۴۱].

declarative + optimizer — batch شبیه MPP (عملکرد comparable)؛ extensibility کد arbitrary و فرمت دلخواه — flexibility.

تخصص‌سازی برای domain مختلف

extensibility کد arbitrary مفید، اما pattern رایج تکرار — reusable building block. MPP سنتی BI و reporting؛ batch domain بیشتر.

domain مهم: statistical/numerical برای ML — Mahout روی MapReduce/Spark/Flink؛ MADlib در HAWQ [۵۴].

الگوریتم spatial مثل k-nearest neighbors [۸۰]. approximate search برای genome [۸۱].

engine batch domain فزاینده — built-in + declarative operator و MPP programmable — در نهایت همه سیستم storage و پردازش داده.

جمع‌بندی

در این فصل پردازش دسته‌ای را بررسی کردیم. از awk، grep و sort شروع کردیم و دیدیم فلسفه Unix به MapReduce و dataflow engine منتقل شد. اصول: ورودی immutable، خروجی ورودی برنامه بعد (ناشناخته)، حل مسائل پیچیده با compose ابزار کوچک «یک کار خوب».

در Unix رابط یکنواخت فایل و pipe؛ در MapReduce filesystem توزیع‌شده. dataflow engine pipe-like transport برای avoid materialization حالت میانی — ورودی/خروجی نهایی job معمولاً HDFS.

دو مسئله اصلی framework batch توزیع‌شده:

Partitioning در MapReduce mapper بر block فایل ورودی partition. خروجی mapper repartition، sort و merge به تعداد reducer. هدف: همه داده مرتبط — مثلاً همان key — در یک جا. dataflow post-MapReduce sort فقط وقتی لازم؛ partitioning broadly مشابه.

Fault tolerance MapReduce frequent write دیسک — recover task fail بدون restart کل job اما کندتر بدون fault. dataflow materialization کمتر، حافظه بیشتر — recompute بیشتر در fail. operator deterministic recompute کمتر.

join algorithm MapReduce — در MPP و dataflow داخلی — partitioned algorithm:

Sort-merge join هر ورودی mapper با join key. partition، sort، merge — همه رکورد یک key به یک reducer. reducer join logic.

Broadcast hash join یک ورودی کوچک — partition نشده، hash table در هر mapper ورودی بزرگ.

Partitioned hash join هر دو ورودی یکسان partition (همان key، hash، تعداد partition) — hash table per partition.

مدل برنامه‌نویسی batch عمداً محدود: callback stateless بدون side effect visible جز خروجی designated. فریم‌ورک crash و شبکه را پنهان — task retry safe، خروجی task fail discard. چند task موفق برای partition — فقط یکی output visible.

کد batch از fault tolerance در framework — خروجی نهایی مثل بدون fault حتی با retry. semantics قوی‌تر از سرویس آنلاین با side effect نوشتن پایگاه.

ویژگی batch: ورودی می‌خواند خروجی تولید می‌کند بدون تغییر ورودی — خروجی از ورودی derive. ورودی bounded: اندازه known ثابت (log در زمان مشخص، snapshot پایگاه). job می‌داند کی تمام خواندن و eventually complete.

فصل ۱۱ stream processing — ورودی unbounded؛ job never complete. stream و batch شباهت دارند اما unbounded assumption زیاد عوض می‌کند.

منابع

[1] Jeffrey Dean and Sanjay Ghemawat: "MapReduce: Simplified Data Processing on Large Clusters," at 6th USENIX Symposium on Operating System Design and Implementation (OSDI), December 2004.

[2] Joel Spolsky: "The Perils of JavaSchools," joelonsoftware.com, December 25, 2005.

[3] Shivnath Babu and Herodotos Herodotou: "Massively Parallel Databases and MapReduce Systems," Foundations and Trends in Databases, volume 5, number 1, pages 1–104, November 2013. doi:10.1561/1900000036

[4] David J. DeWitt and Michael Stonebraker: "MapReduce: A Major Step Backwards," originally published at databasecolumn.vertica.com, January 17, 2008.

[5] Henry Robinson: "The Elephant Was a Trojan Horse: On the Death of MapReduce at Google," the-paper-trail.org, June 25, 2014.

[6] "The Hollerith Machine," United States Census Bureau, census.gov.

[7] "IBM 82, 83, and 84 Sorters Reference Manual," Edition A24-1034-1, International Business Machines Corporation, July 1962.

[8] Adam Drake: "Command-Line Tools Can Be 235x Faster than Your Hadoop Cluster," aadrake.com, January 25, 2014.

[9] "GNU Coreutils 8.23 Documentation," Free Software Foundation, Inc., 2014.

[10] Martin Kleppmann: "Kafka, Samza, and the Unix Philosophy of Distributed Data," martin.kleppmann.com, August 5, 2015.

[11] Doug McIlroy: Internal Bell Labs memo, October 1964. Cited in: Dennis M. Richie: "Advice from Doug McIlroy," cm.bell-labs.com.

[12] M. D. McIlroy, E. N. Pinson, and B. A. Tague: "UNIX Time-Sharing System: Foreword," The Bell System Technical Journal, volume 57, number 6, pages 1899–1904, July 1978.

[13] Eric S. Raymond: The Art of UNIX Programming. Addison-Wesley, 2003. ISBN: 978-0-13-142901-7

[14] Ronald Duncan: "Text File Formats – ASCII Delimited Text – Not CSV or TAB Delimited Text," ronaldduncan.wordpress.com, October 31, 2009.

[15] Alan Kay: "Is 'Software Engineering' an Oxymoron?," tinlizzie.org.

[16] Martin Fowler: "InversionOfControl," martinfowler.com, June 26, 2005.

[17] Daniel J. Bernstein: "Two File Descriptors for Sockets," cr.yp.to.

[18] Rob Pike and Dennis M. Ritchie: "The Styx Architecture for Distributed Systems," Bell Labs Technical Journal, volume 4, number 2, pages 146–152, April 1999.

[19] Sanjay Ghemawat, Howard Gobioff, and Shun-Tak Leung: "The Google File System," at 19th ACM Symposium on Operating Systems Principles (SOSP), October 2003. doi:10.1145/945445.945450

[20] Michael Ovsiannikov, Silvius Rus, Damian Reeves, et al.: "The Quantcast File System," Proceedings of the VLDB Endowment, volume 6, number 11, pages 1092–1101, August 2013. doi:10.14778/2536222.2536234

[21] "OpenStack Swift 2.6.1 Developer Documentation," OpenStack Foundation, docs.openstack.org, March 2016.

[22] Zhe Zhang, Andrew Wang, Kai Zheng, et al.: "Introduction to HDFS Erasure Coding in Apache Hadoop," blog.cloudera.com, September 23, 2015.

[23] Peter Cnudde: "Hadoop Turns 10," yahoohadoop.tumblr.com, February 5, 2016.

[24] Eric Baldeschwieler: "Thinking About the HDFS vs. Other Storage Technologies," hortonworks.com, July 25, 2012.

[25] Brendan Gregg: "Manta: Unix Meets Map Reduce," dtrace.org, June 25, 2013.

[26] Tom White: Hadoop: The Definitive Guide, 4th edition. O'Reilly Media, 2015. ISBN: 978-1-491-90163-2

[27] Jim N. Gray: "Distributed Computing Economics," Microsoft Research Tech Report MSR-TR-2003-24, March 2003.

[28] Márton Trencséni: "Luigi vs Airflow vs Pinball," bytepawn.com, February 6, 2016.

[29] Roshan Sumbaly, Jay Kreps, and Sam Shah: "The 'Big Data' Ecosystem at LinkedIn," at ACM International Conference on Management of Data (SIGMOD), July 2013. doi:10.1145/2463676.2463707

[30] Alan F. Gates, Olga Natkovich, Shubham Chopra, et al.: "Building a High-Level Dataflow System on Top of Map-Reduce: The Pig Experience," at 35th International Conference on Very Large Data Bases (VLDB), August 2009.

[31] Ashish Thusoo, Joydeep Sen Sarma, Namit Jain, et al.: "Hive – A Petabyte Scale Data Warehouse Using Hadoop," at 26th IEEE International Conference on Data Engineering (ICDE), March 2010. doi:10.1109/ICDE.2010.5447738

[32] "Cascading 3.0 User Guide," Concurrent, Inc., docs.cascading.org, January 2016.

[33] "Apache Crunch User Guide," Apache Software Foundation, crunch.apache.org.

[34] Craig Chambers, Ashish Raniwala, Frances Perry, et al.: "FlumeJava: Easy, Efficient Data-Parallel Pipelines," at 31st ACM SIGPLAN Conference on Programming Language Design and Implementation (PLDI), June 2010. doi:10.1145/1806596.1806638

[35] Jay Kreps: "Why Local State is a Fundamental Primitive in Stream Processing," oreilly.com, July 31, 2014.

[36] Martin Kleppmann: "Rethinking Caching in Web Apps," martin.kleppmann.com, October 1, 2012.

[37] Mark Grover, Ted Malaska, Jonathan Seidman, and Gwen Shapira: Hadoop Application Architectures. O'Reilly Media, 2015. ISBN: 978-1-491-90004-8

[38] Philippe Ajoux, Nathan Bronson, Sanjeev Kumar, et al.: "Challenges to Adopting Stronger Consistency at Scale," at 15th USENIX Workshop on Hot Topics in Operating Systems (HotOS), May 2015.

[39] Sriranjan Manjunath: "Skewed Join," wiki.apache.org, 2009.

[40] David J. DeWitt, Jeffrey F. Naughton, Donovan A. Schneider, and S. Seshadri: "Practical Skew Handling in Parallel Joins," at 18th International Conference on Very Large Data Bases (VLDB), August 1992.

[41] Marcel Kornacker, Alexander Behm, Victor Bittorf, et al.: "Impala: A Modern, Open-Source SQL Engine for Hadoop," at 7th Biennial Conference on Innovative Data Systems Research (CIDR), January 2015.

[42] Matthieu Monsch: "Open-Sourcing PalDB, a Lightweight Companion for Storing Side Data," engineering.linkedin.com, October 26, 2015.

[43] Daniel Peng and Frank Dabek: "Large-Scale Incremental Processing Using Distributed Transactions and Notifications," at 9th USENIX conference on Operating Systems Design and Implementation (OSDI), October 2010.

[44] "Cloudera Search User Guide," Cloudera, Inc., September 2015.

[45] Lili Wu, Sam Shah, Sean Choi, et al.: "The Browsemaps: Collaborative Filtering at LinkedIn," at 6th Workshop on Recommender Systems and the Social Web (RSWeb), October 2014.

[46] Roshan Sumbaly, Jay Kreps, Lei Gao, et al.: "Serving Large-Scale Batch Computed Data with Project Voldemort," at 10th USENIX Conference on File and Storage Technologies (FAST), February 2012.

[47] Varun Sharma: "Open-Sourcing Terrapin: A Serving System for Batch Generated Data," engineering.pinterest.com, September 14, 2015.

[48] Nathan Marz: "ElephantDB," slideshare.net, May 30, 2011.

[49] Jean-Daniel (JD) Cryans: "How-to: Use HBase Bulk Loading, and Why," blog.cloudera.com, September 27, 2013.

[50] Nathan Marz: "How to Beat the CAP Theorem," nathanmarz.com, October 13, 2011.

[51] Molly Bartlett Dishman and Martin Fowler: "Agile Architecture," at O'Reilly Software Architecture Conference, March 2015.

[52] David J. DeWitt and Jim N. Gray: "Parallel Database Systems: The Future of High Performance Database Systems," Communications of the ACM, volume 35, number 6, pages 85–98, June 1992. doi:10.1145/129888.129894

[53] Jay Kreps: "But the multi-tenancy thing is actually really really hard," tweetstorm, twitter.com, October 31, 2014.

[54] Jeffrey Cohen, Brian Dolan, Mark Dunlap, et al.: "MAD Skills: New Analysis Practices for Big Data," Proceedings of the VLDB Endowment, volume 2, number 2, pages 1481–1492, August 2009. doi:10.14778/1687553.1687576

[55] Ignacio Terrizzano, Peter Schwarz, Mary Roth, and John E. Colino: "Data Wrangling: The Challenging Journey from the Wild to the Lake," at 7th Biennial Conference on Innovative Data Systems Research (CIDR), January 2015.

[56] Paige Roberts: "To Schema on Read or to Schema on Write, That Is the Hadoop Data Lake Question," adaptivesystemsinc.com, July 2, 2015.

[57] Bobby Johnson and Joseph Adler: "The Sushi Principle: Raw Data Is Better," at Strata+Hadoop World, February 2015.

[58] Vinod Kumar Vavilapalli, Arun C. Murthy, Chris Douglas, et al.: "Apache Hadoop YARN: Yet Another Resource Negotiator," at 4th ACM Symposium on Cloud Computing (SoCC), October 2013. doi:10.1145/2523616.2523633

[59] Abhishek Verma, Luis Pedrosa, Madhukar Korupolu, et al.: "Large-Scale Cluster Management at Google with Borg," at 10th European Conference on Computer Systems (EuroSys), April 2015. doi:10.1145/2741948.2741964

[60] Malte Schwarzkopf: "The Evolution of Cluster Scheduler Architectures," firmament.io, March 9, 2016.

[61] Matei Zaharia, Mosharaf Chowdhury, Tathagata Das, et al.: "Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing," at 9th USENIX Symposium on Networked Systems Design and Implementation (NSDI), April 2012.

[62] Holden Karau, Andy Konwinski, Patrick Wendell, and Matei Zaharia: Learning Spark. O'Reilly Media, 2015. ISBN: 978-1-449-35904-1

[63] Bikas Saha and Hitesh Shah: "Apache Tez: Accelerating Hadoop Query Processing," at Hadoop Summit, June 2014.

[64] Bikas Saha, Hitesh Shah, Siddharth Seth, et al.: "Apache Tez: A Unifying Framework for Modeling and Building Data Processing Applications," at ACM International Conference on Management of Data (SIGMOD), June 2015. doi:10.1145/2723372.2742790

[65] Kostas Tzoumas: "Apache Flink: API, Runtime, and Project Roadmap," slideshare.net, January 14, 2015.

[66] Alexander Alexandrov, Rico Bergmann, Stephan Ewen, et al.: "The Stratosphere Platform for Big Data Analytics," The VLDB Journal, volume 23, number 6, pages 939–964, May 2014. doi:10.1007/s00778-014-0357-y

[67] Michael Isard, Mihai Budiu, Yuan Yu, et al.: "Dryad: Distributed Data-Parallel Programs from Sequential Building Blocks," at European Conference on Computer Systems (EuroSys), March 2007. doi:10.1145/1272996.1273005

[68] Daniel Warneke and Odej Kao: "Nephele: Efficient Parallel Data Processing in the Cloud," at 2nd Workshop on Many-Task Computing on Grids and Supercomputers (MTAGS), November 2009. doi:10.1145/1646468.1646476

[69] Lawrence Page, Sergey Brin, Rajeev Motwani, and Terry Winograd: "The PageRank Citation Ranking: Bringing Order to the Web," Stanford InfoLab Technical Report 422, 1999.

[70] Leslie G. Valiant: "A Bridging Model for Parallel Computation," Communications of the ACM, volume 33, number 8, pages 103–111, August 1990. doi:10.1145/79173.79181

[71] Stephan Ewen, Kostas Tzoumas, Moritz Kaufmann, and Volker Markl: "Spinning Fast Iterative Data Flows," Proceedings of the VLDB Endowment, volume 5, number 11, pages 1268-1279, July 2012. doi:10.14778/2350229.2350245

[72] Grzegorz Malewicz, Matthew H. Austern, Aart J. C. Bik, et al.: "Pregel: A System for Large-Scale Graph Processing," at ACM International Conference on Management of Data (SIGMOD), June 2010. doi:10.1145/1807167.1807184

[73] Frank McSherry, Michael Isard, and Derek G. Murray: "Scalability! But at What COST?," at 15th USENIX Workshop on Hot Topics in Operating Systems (HotOS), May 2015.

[74] Ionel Gog, Malte Schwarzkopf, Natacha Crooks, et al.: "Musketeer: All for One, One for All in Data Processing Systems," at 10th European Conference on Computer Systems (EuroSys), April 2015. doi:10.1145/2741948.2741968

[75] Aapo Kyrola, Guy Blelloch, and Carlos Guestrin: "GraphChi: Large-Scale Graph Computation on Just a PC," at 10th USENIX Symposium on Operating Systems Design and Implementation (OSDI), October 2012.

[76] Andrew Lenharth, Donald Nguyen, and Keshav Pingali: "Parallel Graph Analytics," Communications of the ACM, volume 59, number 5, pages 78–87, May 2016. doi:10.1145/2901919

[77] Fabian Hüske: "Peeking into Apache Flink's Engine Room," flink.apache.org, March 13, 2015.

[78] Mostafa Mokhtar: "Hive 0.14 Cost Based Optimizer (CBO) Technical Overview," hortonworks.com, March 2, 2015.

[79] Michael Armbrust, Reynold S Xin, Cheng Lian, et al.: "Spark SQL: Relational Data Processing in Spark," at ACM International Conference on Management of Data (SIGMOD), June 2015. doi:10.1145/2723372.2742797

[80] Daniel Blazevski: "Planting Quadtrees for Apache Flink," insightdataengineering.com, March 25, 2016.

[81] Tom White: "Genome Analysis Toolkit: Now Using Apache Spark for Data Processing," blog.cloudera.com, April 6, 2016.