ב-Managed Service for Apache Spark Enhanced Flexibility Mode (EFM), הנתונים של פעולת ה-shuffle מנוהלים כדי לצמצם את העיכובים בהתקדמות העבודה שנגרמים כתוצאה מהסרת צמתים מאשכול פעיל. EFM מעביר נתוני ערבוב על ידי כתיבת נתונים לעובדים ראשיים. העובדים שולפים מהצמתים המרוחקים האלה במהלך שלב הצמצום. המצב הזה זמין רק לעבודות Spark.
מכיוון ש-EFM לא מאחסן נתוני ערבוב ביניים בעובדים משניים, הוא מתאים במיוחד לאשכולות שמשתמשים במכונות וירטואליות שניתן לקטוע את הפעולה שלהן או שמבצעים שינוי גודל אוטומטי רק של קבוצת העובדים המשניים.
EFM נתמך ב-Managed Service for Apache Spark
2.0.31+, 2.1.6+, 2.2+ ובגרסאות מאוחרות יותר של תמונות.
- משימות של Apache Hadoop YARN שלא תומכות בהעברה של AppMaster עלולות להיכשל במצב גמישות משופרת (ראו מתי צריך לחכות לסיום של AppMasters).
- לא מומלץ להשתמש במצב גמישות משופר:
- באשכול שיש בו רק עובדים ראשיים
- במשימות סטרימינג, כי יכולות לעבור עד 30 דקות אחרי השלמת המשימה עד שנתוני הביניים של ה-shuffle ינוקו.
- באשכול שמריץ מחברות, כי יכול להיות שנתוני הערבוב לא ינוקו במהלך חיי הסשן.
- כשמשימות Spark מופעלות באשכול שבו מופעלת האפשרות 'צמצום הדרגתי של כמות השרתים'. צמצום הדרגתי של כמות השרתים ו-EFM יכולים לפעול למטרות מנוגדות, כי מנגנון צמצום הדרגתי של כמות השרתים של YARN שומר את הצמתים בהשבתה עד שכל האפליקציות הרלוונטיות מסתיימות.
- באשכול שמריץ משימות Spark ומשימות אחרות.
- מצב גמישות משופר לא נתמך:
- כשההגדרה 'שינוי גודל אוטומטי של העובד הראשי' מופעלת. ברוב המקרים, עובדים ראשיים ימשיכו לאחסן נתוני ערבוב שלא מועברים אוטומטית. הקטנת קבוצת ה-primary worker מבטלת את היתרונות של EFM.
שימוש במצב גמישות משופר
הגמישות המשופרת מופעלת כשיוצרים אוסף על ידי הגדרת dataproc:efm.spark.shuffle מאפיין האוסף לערך primary-worker.
לדוגמה:
gcloud dataproc clusters create cluster-name \ --region=region \ --properties=dataproc:efm.spark.shuffle=primary-worker \ other flags ...
דוגמה ל-Apache Spark
- מריצים משימה של ספירת מילים מול טקסט ציבורי של שייקספיר באמצעות קובץ ה-jar של דוגמאות Spark באשכול EFM.
gcloud dataproc jobs submit spark \ --cluster=cluster-name \ --region=region \ --jars=file:///usr/lib/spark/examples/jars/spark-examples.jar \ --class=org.apache.spark.examples.JavaWordCount \ -- gs://apache-beam-samples/shakespeare/macbeth.txt
הגדרת כונני SSD מקומיים
מכיוון ש-EFM כותב נתוני ערבוב ביניים לדיסקים שמצורפים למכונות וירטואליות, הוא נהנה מרוחב הפס הנוסף וממספר פעולות הקלט/פלט בשנייה שמסופקים על ידי כונני SSD מקומיים. כדי להקל על הקצאת משאבים, כשמגדירים מכונות עובד ראשיות, מומלץ לכוון להקצאה של מחיצת SSD מקומית אחת לכל 4 ליבות vCPU.
כדי לצרף כונני SSD מקומיים, מעבירים את הדגל --num-worker-local-ssds לפקודה gcloud Managed Service for Apache Spark clusters create.
בדרך כלל, לא תצטרכו כונני SSD מקומיים בעובדים משניים.
בדרך כלל אין חשיבות רבה להוספת כונני SSD מקומיים לעובדים המשניים של אשכול (באמצעות הדגל --num-secondary-worker-local-ssds), כי העובדים המשניים לא כותבים נתונים של ערבוב באופן מקומי.
עם זאת, מכיוון ש-SSD מקומיים משפרים את הביצועים של הדיסק המקומי, יכול להיות שתחליטו להוסיף SSD מקומיים לעובדים משניים אם אתם מצפים שעבודות יהיו מוגבלות על ידי קלט/פלט בגלל שימוש בדיסק מקומי: העבודה שלכם משתמשת בדיסק מקומי משמעותי עבור שטח אחסון זמני, או שהמחיצות שלכם גדולות מדי מכדי להיכנס לזיכרון ויעברו לדיסק.
יחס עובדים משני
מכיוון שעובדים משניים כותבים את נתוני ה-shuffle שלהם לעובדים ראשיים, האשכול צריך להכיל מספר מספיק של עובדים ראשיים עם מספיק משאבי CPU, זיכרון ודיסק כדי להתאים לעומס ה-shuffle של העבודה. כדי למנוע את שינוי הגודל של הקבוצה הראשית באשכולות עם שינוי גודל אוטומטי, וכך למנוע התנהגות לא רצויה, צריך להגדיר את minInstances לערך maxInstances במדיניות שינוי הגודל האוטומטי של קבוצת העובדים הראשית.
אם יש לכם יחס גבוה בין העובדים המשניים לבין העובדים הראשיים (לדוגמה, 10:1), כדאי לעקוב אחרי השימוש ב-CPU, ברשת ובדיסק של העובדים הראשיים כדי לקבוע אם יש עומס יתר. לשם כך:
נכנסים לדף VM instances במסוףGoogle Cloud .
לוחצים על תיבת הסימון שמימין ל-primary worker.
לוחצים על הכרטיסייה MONITORING (מעקב) כדי לראות את נתוני השימוש במעבד של העובד הראשי, את נתוני ה-IOPS של הדיסק, את נתוני הבייטים ברשת ומדדים אחרים.
אם עומס העבודה על העובדים הראשיים גבוה מדי, כדאי להגדיל את מספר העובדים הראשיים באופן ידני.
שינוי הגודל של קבוצת העובדים הראשית
אפשר להגדיל את מספר העובדים בקבוצת העובדים הראשית בלי בעיות, אבל הקטנת מספר העובדים בקבוצת העובדים הראשית עלולה להשפיע לרעה על התקדמות העבודה. פעולות שמקטינות את קבוצת ה-primary worker צריכות להשתמש בצמצום הדרגתי של כמות השרתים, שמופעל על ידי הגדרת הדגל --graceful-decommission-timeout.
אשכולות עם התאמה אוטומטית לעומס: המידרוג של קבוצת העובדים הראשית מושבת באשכולות EFM עם מדיניות של התאמה אוטומטית לעומס. כדי לשנות את הגודל של קבוצת העובדים הראשית באשכול עם מידרוג אוטומטי:
משביתים את שינוי הגודל האוטומטי.
gcloud dataproc clusters update \ --cluster=cluster-name \ --region=region \ --disable-autoscaling
שינוי הגודל של הקבוצה הראשית.
gcloud dataproc clusters update \ --cluster=cluster-name \ --region=region \ --num-workers=num-primary-workers \ --graceful-decommission-timeout=graceful-decommission-timeout # (if downscaling)
הפעלה מחדש של שינוי הגודל האוטומטי:
gcloud dataproc clusters update \ --cluster=cluster-name \ --region=region \ --autoscaling-policy=autoscaling-policy
מעקב אחר השימוש בדיסק של העובד הראשי
לעובדים הראשיים צריך להיות מספיק מקום בדיסק בשביל נתוני הערבוב של האשכול.
אפשר לעקוב אחרי זה באופן עקיף באמצעות המדד remaining HDFS capacity.
כשהכונן המקומי מתמלא, אין יותר מקום פנוי ל-HDFS, והקיבולת שנותרה קטנה יותר.
כברירת מחדל, אם השימוש בדיסק המקומי של עובד ראשי עולה על 90% מהקיבולת, הצומת יסומן כ-UNHEALTHY בממשק המשתמש של צומת YARN. אם נתקלים בבעיות בקיבולת הדיסק, אפשר למחוק נתונים שלא בשימוש מ-HDFS או להגדיל את מספר העובדים במאגר העובדים הראשי.
הגדרה מתקדמת
חלוקה למחיצות (partitioning) ומקביליות
כששולחים משימת Spark, צריך להגדיר רמת חלוקה מתאימה. כדי להחליט על מספר מחיצות הקלט והפלט בשלב של ערבוב נתונים, צריך להתפשר בין מאפייני ביצועים שונים. מומלץ להתנסות עם ערכים שמתאימים למבנה התפקידים שלכם.
מחיצות קלט
חלוקת הקלט למחיצות ב-Spark וב-MapReduce נקבעת לפי מערך נתוני הקלט. כשקוראים קבצים מ-Cloud Storage, כל משימה מעבדת נתונים בגודל של בערך 'גודל בלוק' אחד.
במשימות Spark SQL, הגודל המקסימלי של המחיצה נקבע על ידי
spark.sql.files.maxPartitionBytes. מומלץ להגדיל את הנפח ל-1GB:spark.sql.files.maxPartitionBytes=1073741824.ב-RDD של Spark, גודל המחיצה נשלט בדרך כלל באמצעות
fs.gs.block.size, שערך ברירת המחדל שלו הוא 128MB. כדאי להגדיל אותו ל-1GB. לדוגמה:--properties spark.hadoop.fs.gs.block.size=1073741824
מחיצות פלט
מספר המשימות בשלבים הבאים נקבע על ידי כמה מאפיינים. במשימות גדולות יותר שמעבדות יותר מ-1TB, כדאי להקצות לפחות 1GB לכל מחיצה.
ב-Spark SQL, מספר מחיצות הפלט נשלט על ידי
spark.sql.shuffle.partitions.במשימות Spark שמשתמשות ב-RDD API, אפשר לציין את מספר מחיצות הפלט או להגדיר את
spark.default.parallelism.
כוונון מיון הנתונים (shuffle) עבור primary worker
המאפיין המשמעותי ביותר הוא --properties yarn:spark.shuffle.io.serverThreads=<num-threads>.
שימו לב שזוהי מאפיין YARN ברמת האשכול, כי שרת ה-shuffle של Spark פועל כחלק מ-Node Manager. ברירת המחדל היא פי שניים (2x) ממספר ליבות המעבד במכונה (לדוגמה, 16 תהליכים ב-n1-highmem-8). אם הערך של 'ערבוב קריאה חסום זמן' גדול משנייה אחת, והעובדים הראשיים לא הגיעו למגבלות של הרשת, המעבד או הדיסק, כדאי להגדיל את מספר השרשורים של שרת הערבוב.
בסוגי מכונות גדולים יותר, כדאי להגדיל את הערך של spark.shuffle.io.numConnectionsPerPeer, שמוגדר כברירת מחדל ל-1. (לדוגמה, אפשר להגדיר 5 חיבורים לכל צמד מארחים).
הגדלת מספר הניסיונות החוזרים
אפשר להגדיר את מספר הניסיונות המקסימלי שמותר למאסטרים של אפליקציות, למשימות ולשלבים באמצעות הגדרת המאפיינים הבאים:
yarn:yarn.resourcemanager.am.max-attempts spark:spark.task.maxFailures spark:spark.stage.maxConsecutiveAttempts
מכיוון שמשימות ואפליקציות ראשיות מסתיימות בתדירות גבוהה יותר באשכולות שמשתמשים בהרבה מכונות VM זמניות או בהתאמה אוטומטית לעומס ללא צמצום הדרגתי של כמות השרתים, הגדלת הערכים של המאפיינים הקודמים באשכולות האלה יכולה לעזור (שימו לב שהשימוש ב-EFM עם Spark וצמצום הדרגתי של כמות השרתים לא אפשרי).
צמצום הדרגתי של כמות השרתים ב-YARN באשכולות EFM
אפשר להשתמש ב-YARN Graceful Decommissioning כדי להסיר צמתים במהירות עם השפעה מינימלית על אפליקציות שפועלות. באשכולות עם התאמה אוטומטית לעומס, אפשר להגדיר את זמן קצוב לתפוגה של צמצום הדרגתי של כמות השרתים ב-AutoscalingPolicy שמצורף לאשכול EFM.
שיפורים ב-EFM לצמצום הדרגתי של כמות השרתים
מכיוון שהנתונים הזמניים מאוחסנים במערכת קבצים מבוזרת, אפשר להסיר צמתים מאשכול EFM ברגע שכל הקונטיינרים שפועלים בצמתים האלה סיימו את הפעולה. לעומת זאת, בצבי אשכולות רגילים של Managed Service for Apache Spark, הצמתים לא מוסרים עד שהאפליקציה מסיימת את הפעולה.
הסרת צומת לא ממתינה לסיום של אפליקציות ראשיות שפועלות בצומת. כשהקונטיינר הראשי של האפליקציה מסתיים, הוא מתוזמן מחדש בצומת אחר שלא מוצא משימוש. התקדמות המשימה לא אובדת: המאסטר החדש של האפליקציה משחזר במהירות את המצב מהמאסטר הקודם של האפליקציה על ידי קריאת היסטוריית המשימות.