מערכות מבוזרות - מבט עדכני

בפוסט הקודם, "בעיות של מערכות מבו..זרות", טענתי בטרוניה שבחומר הכתוב על מערכות מבוזרות - כמעט ולא מתארים מערכות מבוזרות בנות-זמננו.

קיטרתי - גם אני לא עשיתי זאת בעצמי (כמעט).

בפוסט זה ארצה לגעת בכמה מערכות מבוזרות - המאפשרות לפתור בעיות מבוזרות מבלי להיכנס לפרטים של "בחירת מנהיג" או "יצירת קונצנזוס". בעיות רבות של עיבוד מבוזר - חוזרות על עצמן, וחבל להתחיל לפתור אותן מהתחלה.

ישנם שני סוגים של מערכות מבוזרות נפוצות שאני יכול לחשוב עליהן:

• מוצרים: אפליקציית לקוח של טורנט, Tor (רשת פרוקסי לאנונימיות), או משחק Massively Multi-player Online.
• Frameworks: המאפשרים פתרון בעיות מבוזרות ב"רמת הפשטה גבוהה".

יש משהו מאוד קוסם בסקירה כיצד מוצרים מבוזרים עובדים - זה יכול להיות ממש מעניין!

האמת היא שיש לי (ובכלל בקהילה) - ידע מוגבל על הנושא. (הנה פוסט נחמד על Tor)

בסקירה של קטגורית Frameworks יש משהו יותר מעשי: סביר יותר שתבחרו Framework לעבוד בו - מאשר לכתוב מערת מבוזרת מ scratch, תוך כדי שאתם מושפעים ממימוש של מערכות אחרות.

כמו כן - הידע עליהן זמין ונגיש. לאחרונה התחלנו לעבוד עם Hadoop - מה שעוזרת לי לשלב את תוכן הפוסט עם העבודה השוטפת.

בסופו של דבר, גם ה Frameworks וגם המוצרים המבוזרים שמתוארים בפוסט עושים שימוש נרחב במנגנונים ("פרימיטיביים") של מערכות מבוזרות שתיארתי בפוסט הקודם: Multi-cast ורפליקציה, בחירת מנהיג, השגת קונצנזוס ועוד. חלק מכובד מהמערכות שאתאר בפוסט זה משתמש ב Zookeeper, מוצר שהזכרתי בפוסט הקודם - בכדי לבצע את הפרימיטיביים המבוזרים הללו.

תוך כדי כתיבת הפוסט - נוכחתי שאני מתמקד בקטגוריה מאוד ספציפית של מוצרים מבוזרים: מוצרי Big Data.

זה בסדר.

מוצרי Big Data לרוב מחשבים... הרבה נתונים. כ"כ הרבה נתונים שלא ניתן לאחסן במחשב אחד. לרוב גם לא בעשרה.

כדי לטפל בכמויות הנתונים (Volume) או במהירות שלהם (Velocity) - זקוקים למערכת מבוזרת. אותם מוצרים (או Frameworks) הם גם במובן אחד "מערכת מבוזרת", וגם "מוצר Big Data". בפוסט - נבחן בעיקר את הפן המבוזר שלהן.

בכל מקרה: כנראה שחלק מכובד מאוד מאנשי התוכנה שמתמודדים עם מערכות מבוזרות - עושים זאת דרך עבודה עם מוצרי Big Data, הפופולריים כ"כ היום.

בקיצור: זהו איננו פוסט על Big Data. יום אחד אולי אקדיש לנושא פוסט ראוי, ואולי גם לא :)

חישוב מבוזר

Scatter - Gather הוא השם המקובל לדפוס חישוב מבוזר, המבוסס על 2 רעיונות:

• חישוב הוא מהיר יותר על מערכת מקומית (גישה לזיכרון מהירה מגישה ברשת, גישה לדיסק מקומי לרוב תהיה מהירה מגישה לדיסק מרוחק) - ולכן נבצע את החישוב היכן שנמצאים הנתונים.
• למחשב יחיד אין דיי שטח אכסון לכל הנתונים של "חישוב גדול", וגם אם היה לו - לא היה לו את כח החישוב המספיק לעבד את כל הנתונים הללו בזמן סביר.

נ.ב. "זמן סביר" הוא יחסי. לא נדיר להיתקל בשאילתות שרצות על מערכות Hadoop במשך שבועות - עד לסיום החישוב.



Map Reduce - הוא היישום הידוע של פרדיגמת Scatter-Gather, שנחשב היום אמנם כבוגר - ומובן היטב.

הרעיון הוא לחלק את החישוב המבוזר לשני שלבים:

• Map - עיבוד נתונים נקודתי על המכונה הלוקאלית (חיפוש / פילטור / עיבוד נתונים וכו')
• Reduce - ביצוע סיכום של החישוב וחיבור התשובות של המחשבים הבודדים לתשובה אחת גדולה.

החלוצים בתחום של Map Reduce מבוזר על פני מספר רב של מחשבים הם חברת גוגל, ופרויקט Apache Hadoop - שהושפע רבות ממאמרים שגוגל פרסמה על מימוש ה Mapreduce שלה. בעשור מאז הושק הפך פרויקט Hadoop לסטנדט בתעשייה, עם קהילה עשירה ורחבה. כאשר הרווחים מתמיכה ב Hadoop הגיעה למאות מיליוני דולרים בשנה - נכנסו לשם גם שחקני ה Enterprise הגדולים (כלומר: EMC, IBM, Oracle, וכו')
.
Google Map-reduce ו Hadoop מבוססים על מערכת קבצים מבוזרת (GFS ו HDFS בהתאמה), ותשתית לביצוע חישוב מבוזר על גבי ה nodes שמחזיקים את הנתונים הנ"ל (ב Hadoop נקראת YARN, בגוגל כנראה פשוט "Mapreduce").

בפועל יש ב Map reduce שלושה שלבים עיקריים: Reduce, Map ו Shuffle (העברת סיכומי הביניים בין ה Mappers ל Reducers - פעולה אותו מספק ה Framework).

דוגמה לבעיית מיון מבוזרת (הניחו שמדובר במיליארדי רשומות) הנפתרת בעזרת MapReduce.
המפתחים כותבים את פונקציות ה Map וה Reduce, ה framework מספק להם את ה Shuffle (העברת הנתונים בין ה mappers ל reducers).

ניתן לכתוב פונקציות "Map" ו "Reduce" בקוד, אך סביר יותר להניח שכאשר אתם מגדירים חישובי Map-reduce רבים - תעדיפו להשתמש ברמת הפשטה גבוהה יותר כגון אלו שמספקים Pig (מן "שפת תכנות" מיוחדת ל Map-Reduce) או Hive (המבוסס על SQL).

כדרך אגב: Pig קיבל את שמו מכיוון שחזירים אוכלים גם צמחים וגם בשר - ו Pig יודע לטפל גם ב structured data וגם ב unstructured data.

השנים עברו, וכיום המודל של חלוקת החישוב המבוזר לשלבי Map ו Reduce  - נחשב מעט מוגבל.

כמו כן, ההסתמכות על מערכת קבצים כבסיס לחישוב היא טובה ל DataSets ענק שלא ניתן להכיל בזיכרון (יש clusters של Hadoop שמנתחים עשרות, ואולי מאות PetaBytes של נתונים), אבל בעשור האחרון הזיכרון הפך לזול (מכונה עם 1TB זכרון היא בהישג יד) - והיכולת לבצע את החישוב בזיכרון יכולה להאיץ את מהירות החישוב בסדר גודל או שניים, ואולי אף יותר (תלוי בחישוב). עובדה זו מקדמת את הפופולריות של הפתרונות לחישוב מבוזר מבוסס-הזיכרון (כמו Storm, או Spark), ומשאירה ל Hadoop את היתרון בעיקר עבור מאגרי הנתונים הבאמת-גדולים (נאמר: 100TB ויותר של נתונים, מספר שבוודאי ישתנה עם הזמן).

המוצרים העיקריים ב Eco-System של Hadoop

דוגמה נוספת לדפוס של Scatter - Gather הוא הדפוס בו עושים שימוש ב Spark. אני לא בטוח שזהו שם רשמי - אך אני מכיר את המינוח "Transform - Action" (כאשר "map" הוא פעולת transform אפשרית, ו "reduce" הוא action אפשרי).

Spark הוא הכוכב העולה של עולם ה Big Data. בניגוד ל Hadoop הוא מאפשר (כחלק מהמוצר) מספר רב יותר של פונקציות מאשר "map" ו "reduce".
במקום פעולת ה map ניתן להשתמש בסמנטיקות כגון map, filter, sample, union, join, partitionby ועוד...
במקום פעולת ה reduce ניתן להשתמש בסמנטיקות כגון reduce, collect, count, first, foreach, saveAsXxx ועוד.

מעבר לגמישות הזו - לא חייבים לתאר את הבעיה כ-2 שלבי חישוב, אלא ניתן להגדיר גרף חישוב גמיש כמעט לחלוטין.
עניין זה חשוב - כי יש בעיות חישוביות שפשוט לא מתמפות יפה ל 2 השלבים הברורים האלו. למשל: כאשר יש צורך להשתמש באותם נתוני-מקור כמה פעמים במהלך תהליך החישוב.

המוצרים ב Eco-System ה מתהווה של Spark (חלקם מקבילים לכלים קיימים של Hadoop). מקור: databricks

ל Spark יש עוד שתי תכונות חשובות:

• הוא מחזיק (סוג של Cache) את הנתונים בזיכרון - אליו הגישה מהירה בסדרי גודל מאשר בדיסק.
• הוא מרכז את רוב שרשרת החישוב באותו node ומצמצם את הצורך להעביר נתונים בין nodes ב cluster (גם ל Hadoop יש כלים שעוזרים לצמצם העברות של נתונים בין nodes, אך Map-reduce פחות מוצלח בכך באופן טבעי)

פרטים נוספים:

• ה "agent" של Spark רץ במקביל למערכת ה storage המבוזר, תהיה זו Hadoop או Cassandra (או סתם קבצים בפורמט נתמך) - ושואב ממנו את הנתונים הגולמיים לחישוב.
• ה Agent (נקרא node - אני נמנע משם זה בכדי למנוע בלבול) מקבל Job (לצורך העניין: קטע קוד) לבצע חישוב מסוים על הנתונים. יש כבר ב Agent קוד של פונקציות שימושיות (filter, sort, join, וכו' - לפעולות "map" או count, reduce, collect - לפעולות "reduce").
• ה Agent מחלק את המשימה ל stages ובונה גרף חישוב (על בסיס הידע כיצד מאורגנים הנתונים ב cluster) - ושולח tasks חישוביים ל Agents ב cluster.

ל Spark יש לא מעט נקודות אינטגרציה עם Hadoop - כך שהתחרות באמת היא לא בין Hadoop ל Spark (לפחות - עדיין), אלא יותר תחרות בין Spark ל YARN (מנוע ה Map-reduce).

Event Processing

Spark ו Hadoop הם מנועים לחישוב אצוות (batch): יש טריגר מסוים שמתחיל את החישוב - והחישוב מתבצע לאורך זמן מסוים (עדיף שיהיה קצר ככל האפשר, אך הוא לא מוגבל בזמן) - עד שמתקבלת התוצאה.

זן אחר של חישוב הוא חישוב ב Streaming: ביצוע חישובים קטנים רבים, כחלק משטף של אירועים שנכנס למערכת.
בקטגוריה זו ניתן לכלול את Spark Streaming (המבוסס על Spark), אך אתמקד דווקא ב Storm - שנבנה לצורך זה במקור.


Storm

• מערכת Horizontal scalable ו fault-tolerant - כמובן!
• מערכת המספקת תשובות ב low-latency (שלא לומר real-time - כי זה פשוט לא יהיה נכון), תוך כדי שימוש בכוח חישוב מבוזר.
• At-lest-once Processing מה שאומר שהנחת היסוד במערכת המבוזרת היא שחלקים שלה כושלים מדי פעם, ולכן לעתים נעשה אותו חישוב יותר מפעם אחת (לרוב: פעמיים) - בכדי להמשיך לקבל תוצאות גם במקרה של כשל מקומי.

Storm נולדה מתוך פרויקט בטוויטר לחישוב trending analysis ב low-latency.

המבנה של Storm Computation Cluster הוא סוג של Pipes & Filters, כאשר יש:

• Stream - רצף של tuples של נתונים.
• Spout - נקודת אינטגרציה של המערכת, המזרימה נתונים ל streams.
• Bolt - נקודת חישוב ("Filter") בגרף החישוב ("הטופולוגיה").
  על מערכת Storm מריצים טופולוגיות חישוב רבות במקביל - כל אחת מבצעת חישוב אחר.
• Supervisor - הוא node של Storm, שכולל כמה Workers שיכולים להריץ את קוד ה bolts או ה spouts.
• Stream Grouping - האסטרטגיה כיצד לחלק עבודה בין bolts שונים.

בעוד פעולות אצווה סורקות את הנתונים מקצה אחד לקצה שני - וחוזרות עם תשובה ("42"), ב Event Processing כל הזמן נכנסים נתונים, והערך אותו אנו מחשבים (אחד או יותר) - יתעדכן באופן תדיר ("43, לא רגע... עכשיו 47... ועכשיו 40...").

טופולוגיה לדוגמה של Storm. מקור: michael-noll.com

אם העבודה היא חישובית בלבד - כל אחד מה supervisors יוכל להריץ את כל הטופולוגיה (ואז לא צריך להעביר נתונים בין מחשבים במערכת - שזה עדיף). נוכל לבחור, למשל, ב Local Stream Grouping - שמורה ל Storm לשלוח את הנתונים ל Worker על אותה המכונה.

במידה ויש נתונים מסוימים שיש להשתמש בהם לצורך החישוב, והם נמצאים על supervisors מסוימים - יש להגדיר את Stream Grouping כך שיידע לשלוח את הנתונים ל supervisor הנכון. במקרה כזה אולי נרצה לבחור ב Fields Stream Grouping - המורה ל Storm לשלוח את הנתונים למחשב ע"פ שדה מסוים בנתונים (למשל: "ארץ" - כי כל supervisor מחזיק נתונים של ארץ אחרת).

הערכה גסה איזה כלי יכול להתאים לכם - בהינתן בזמני ההמתנה לתשובה שאתם מוכנים לסבול.
* Tez הוא מנוע שכולל שיפורים ל Map-Reduce, שמרביתם מזכירים את הארכיטקטורה של Spark - מלבד הגמישות לתאר תהליכים כיותר מ-2 שלבים עיקריים.

Mahout

ארצה להזכיר בקצרה את Apache Mahout (שהוא חלק מ Hadoop), פירוש השם Mahout הוא "נהג הפיל" - ואכן Mahout רוכבת על הפיל (Hadoop).
זהו כלי שמספק סט של אלגוריתמים שימושיים הממומשים בצורה מבוזרת, חלקם על גבי Map-Reduce וחלקם על גבי Spark - ומאפשר לעבוד ברמת הפשטה גבוהה אפילו יותר.

סוגי הבעיות ש Mahout מכסה (כל אחת - בעזרת כמה אלגוריתמים שונים) הן:

• סינון שיתופי - "אנשים שאהבו את x אהבו גם את y"
• Classification - אפיון "מרחב" הנתונים מסוג מסוים - גם אלו שאין לנו. סוג של חיזוי.
• רגרסיות - הערכת הקשרים בין משתנים שונים.
• Clustering - חלוקה של פריטי מידע לקבוצות - ע"פ פרמטרים מסוימים.
• ועוד

מימושים של האלגוריתמים  הנ"ל תוכלו למצוא בקלות למכונה יחידה, אבל אם יש לכם סט גדול של נתונים, או שאתם רוצים לחשב את האלגוריתמים בצורה תכופה - Mahout יחסוך לכם עבודה רבה!

.

Lambda Architecture

הכלים שהצגנו למעלה - מאפשרים לנו בחירה בין:

• תשובה מדויקת - אטית (למשל: Map-reduce)
• תשובה "לא מדויקת" - מהירה (למשל: Storm או Spark)
  בהנחה שאנו מקריבים דיוק עבור המהירות. למשל: משתמשים בחישוב בנתונים בני חמש דקות - ולא העדכניים ביותר, או חישוב על סמך דגימה חלקית - ולא על סמך כלל הנתונים.

בארגון בעל צרכי מידע - למה לבחור בין האופציות? האם אי אפשר לקבל את שניהם?

מקור: MapR

סוג של שילוב מקובל בין 2 הגישות נקרא "ארכיטקטורת למבדה", ספק אם על שם האות היוונית למבדה (Λ) שאם מטים אותה על הצד מזכירה מאוד את צורת הארכיטקטורה - ואולי בהקשר ל Lambda Calculus המבוססת על פישוט המערכת לפונקציות (בהקשר של תכנות פונקציונלי). הנחות היסוד של הארכיטקטורה הן:

• מייצרים במערכת התפעולית שטף של אירועים - כולם immutable (בתרשים למעלה: "new data stream").
• את האירועים אוספים ומפצלים ל-2 ערוצים:
• ערוץ של שמירה לטווח ארוך - נוסח HDFS או AWS S3, ואז ביצוע שאילתות ארוכות באצווה (Batch).
• ערוץ של עיבוד מיידי והסקת תובנות - נוסח Storm, שבו ההודעות ייזרקו מיד לאחר שיעובדו (הן שמורות לנו מהערוץ הראשון) - הערוץ המהיר.
• את התובנות קצרות הטווח של הערוץ המהיר, ואת התובנות ארוכות הטווח - של ערוץ האצווה מרכזים ל storage שלישי (Service Layer) - לרוב HBase, Impala או RedShift, ממנו נבצע שליפות חזרה למערכת התפעולית.

כמו ארכיטקטורה, "ארכיטקטורת למבדה" היא רק תבנית מקובלת - ונכון יותר יהיה להתאים אותה לצרכים הייחודיים שלכם, ולא להיצמד בכוח לאיזו הגדרה מקובלת..

Apache Kafka

לסיום אני רוצה להזכיר עוד מערכת מבוזרת שהופכת פופולרית - Distributed Message Queue בשם קפקא.

בדומה למערכות הקודמות, קפקא היא מערכת Horizontally scalable ו Fault-tolerant.
בנוסף יש לה מימד של durability - הודעות נשמרות לדיסק ומשוכפלות בין מחשבים שונים.

בקפקא נלקחו בה כמה החלטות תכנוניות בכדי לתמוך בשטף אדיר של אירועים. קפקא מטפלת יפה בקצבים גבוהים מאוד של נתונים, במחיר של latency מסוים בכתיבת ובשליפת הודעות. שמעתי על חבר'ה שחווים latency אף של ל 2-3 שניות.

כלל-אצבע ששמעתי אומר כך: "אם יש לך עד 100,000 הודעות בשנייה - השתמש ב RabbitMQ, אם יש יותר - כדאי לשקול מעבר לקפקא".

כמה הנחות בעולם של קפקא הן:

• הודעות ב Queue הן immutable - בלתי ניתנות לשינוי. ניתן להוסיף הודעות ל queue - ולקרוא אותן, אך לא לעדכן או למחוק אותן. ההודעות יימחקו כחלק מתהליך של ה Queue (למשל: לאחר 24 שעות) - והצרכן צריך לרשום לעצמו ולעקוב אלו הודעות כבר נקראו - כדי לא לקרוא אותן שוב. הן פשוט יושבות שם - כמו על דיסק.
  קפקא מנהלת את ההודעות כקובץ רציף בדיסק - מה שמשפר מאוד את יעילות ה I/O שלה. כל הפעולות נעשות ב Bulks.
• ה Queue נקרא "Topic" ולא מובטח בו סדר מדויק של ההודעות (פרטים בהמשך).
• Broker הוא שרת (או node) במערכת. אם לברוקר אין מספיק דיסק בכדי לאכסן את תוכן ה Topic, או שאין לו מספיק I/O בכדי לכתוב לדיסק את ההודעות בקצב מהיר מספיק - יש לחלק (באחריות ה Admin האנושי) את ה topic על גבי כמה partitions. ההודעות שמתקבלות ישמרו על partitions שונים, וקריאת ההודעות - גם היא תעשה מה partitions השונים.

קפקא עצמה מאוד מרשימה מבחינת יציבות וביצועים - אך זה נעשה במחיר אי-החבאת מורכבויות של מימוש מהמשתמשים שלו:

ה producer צריך לדעת על מספר ה partitions שבשימוש, מכיוון שקפקא מסתמך על כך שהוא עושה בעצמו מן "Client Side Load Balancing" (דומה לרעיון של Client-Side Directory שהסברתי בפוסט הקודם) - ומפזר את ההודעות בין ה partitions השונים (או שכל producer עושה זאת בעצמו - או שהם עושים זאת - כקבוצה).

הפיזור כמובן לא צריך להיות אקראי אלא יכול להיות ע"פ שדה מסוים בהודעה, שדה כמו "ארץ" - שיש לו משמעות עסקית.

ה consumer לא יכול לקרוא את ההודעות ב topic ע"פ הסדר - כי הן מפוזרות בין ה partitions השונים (קריאה נעשית רק מ partition מסוים). אין שום מנגנון בקפקא שמסייע להבין באיזה partition נמצאת ההודעה הבאה. בפועל מה שעושים הוא שמציבים consumer לכל partition - והם מעבדים את ההודעות במקביל, מבלי להתייחס להודעות שב partitions האחרים.

כמו כן - מישהו צריך לעקוב מהי ההודעה האחרונה שנקראה מכל partition: קפקא לא עושה זאת. הדרך המקובלת לעשות זאת היא להיעזר ב Zookeeper לניהול ה state המבוזר הזה (שיהיה יציב, זמין, וכו').

ב AWS Kinesis (וריאציה של Kafka - של אמזון) - מנהלים את ה state המבוזר ב DynamoDB (בסיס נתונים K/V - גם הוא מבוזר).

כל partition ממספר את ההודעות של ה topic באופן בלתי-תלוי. מקור: kafka.apache.org

לא מתאים לכם? - אתם יכולים לסדר את המידע בחזרה בעזרת מערכת נוספת שתעשה את העבודה (למשל: Storm?).

לכל Broker יש רפליקה (אחת או יותר), במבנה של Active-Passive. ה Active משרת את ה consumer - ואם הוא כושל, אחד ה Passives נבחר להיות ה leader - והופך להיות "ה Active החדש".

ה producer הוא האחראי על מדיניות ה consistency ויכול לבחור להחשיב כתיבה ל queue כמוצלחת כאשר ה Leader קיבל אותה, או רק כאשר ה Leader סיים לעשות רפליקציה של הנתונים ל broker נוסף (אחד או יותר - תלוי ברמת הפרנויה שלכם). התוצאה כמובן היא trade-off בין latency ל durability - שניתן לשחק בו בצורה דינאמית.

אפשר להשוות את קפקא למשאית: היא מאוד לא נעימה לצורך יציאה לדייט עם בחורה, או בכדי לקנות משהו בסופר.

אבל אם אתם רוצים להעביר אלפי טונות של מוצר כלשהו - בוודאי שתרצו כמה משאיות שיעשו את העבודה. צי של רכבים קטנים - פשוט לא יעשה עבודה טובה באותה המידה...

סיכום

בפוסט אמנם הצגתי הרבה מוצרי "Big Data" - אך הכוונה הייתה להתמקד באספקטים המבוזרים שלהם.
מערכות מבוזרות מוסיפות סיבוך למערכת - ולכן כדאי להימנע מהן במידת האפשר. כן... ברור לי שלא-מעט מהנדסים מחפשי-אתגר ימצאו את הדרך להשתמש במערכות מבוזרות - גם שאין צורך עסקי לכך.

יש הגיון מסוים, שבמקום לכתוב באסמבלי מערכת מבוזרת בעצמנו - נשתמש במערכות קיימות. המערכות שהצגתי בפוסט זה הן דוגמה טובה לכך. מכאן-לשם, חלק מהמערכות לא יעשו בדיוק את מה שאתם זקוקים לו, וזה רגע טוב לצלול לשכבת הפשטה נמוכה יותר (כמו Zookeeper - או קוד שלכם) - ולפתור נקודתית את הבעיה.


שיהיה בהצלחה!



----

לינקים רלוונטיים:

השוואה בין Spark ל YARN:
http://www.quora.com/What-is-the-difference-between-Apache-Spark-and-Apache-Hadoop-Map-Reduce

מצגת טובה אודות Spark:
http://stanford.edu/~rezab/dao/slides/lec1.pdf