טיפים לשיפור ביצועים של משימות Spark

בסעיפים הבאים מפורטים טיפים שיעזרו לכם לשפר את הביצועים של אפליקציות Spark ב-Managed Service for Apache Spark.

שימוש באשכולות זמניים

כשמשתמשים במודל האשכולות ה "זמניים" של Managed Service for Apache Spark, יוצרים אשכול ייעודי לכל עבודה, וכשהעבודה מסתיימת, מוחקים את האשכול. במודל הארעי, אפשר להתייחס לאחסון ולמחשוב בנפרד, לשמור את נתוני הקלט והפלט של המשימה ב-Cloud Storage או ב-BigQuery ולהשתמש באשכול רק למחשוב ולאחסון נתונים זמני.

בעיות נפוצות באשכולות קבועים

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

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

היתרונות של אשכולות זמניים

היתרונות של אשכולות זמניים:

  • הגדרת הרשאות שונות ב-IAM למשימות שונות באמצעות חשבונות שירות שונים של מכונות וירטואליות ב-Managed Service for Apache Spark.
  • אופטימיזציה של תצורות החומרה והתוכנה של אשכול לכל משימה, ושינוי של תצורות האשכול לפי הצורך.
  • כדאי לשדרג את גרסאות התמונות באשכולות חדשים כדי לקבל את תיקוני האבטחה, תיקוני הבאגים והאופטימיזציות העדכניים ביותר.
  • פתרון בעיות מהיר יותר באשכול מבודד של משימה אחת.
  • כדי לחסוך בעלויות, משלמים רק על זמן הריצה של אשכול זמני, ולא על זמן ההמתנה בין משימות באשכול משותף.

שימוש ב-Spark SQL

‫Spark SQL DataFrame API הוא אופטימיזציה משמעותית של RDD API. אם אתם מבצעים אינטראקציה עם קוד שמשתמש ב-RDD, כדאי לקרוא את הנתונים כ-DataFrame לפני שמעבירים RDD בקוד. בקוד Java או Scala, כדאי להשתמש ב-Dataset API של Spark SQL כקבוצת-על של RDD ו-DataFrames.

שימוש ב-Apache Spark 3

‫Managed Service for Apache Spark 2.0 מתקין את Spark 3, שכולל את התכונות הבאות ושיפורים בביצועים:

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

שימוש בהקצאה דינמית

‫Apache Spark כולל תכונה של הקצאה דינמית שמשנה את מספר המבצעים של Spark בעובדים באשכול. התכונה הזו מאפשרת לעבודות להשתמש באשכול המלא של Managed Service for Apache Spark גם כשהאשכול גדל. התכונה הזו מופעלת כברירת מחדל ב-Managed Service for Apache Spark (spark.dynamicAllocation.enabled מוגדר כ-true). מידע נוסף זמין במאמר בנושא הקצאה דינמית ב-Spark.

שימוש בהתאמה אוטומטית של קנה מידה ב-Managed Service for Apache Spark

‫Managed Service for Apache Spark שינוי גודל אוטומטי מוסיף ומסיר באופן דינמי עובדים של Managed Service for Apache Spark מאשכול, כדי לוודא שלמשימות Spark יש את המשאבים הדרושים להשלמה מהירה.

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

שימוש במצב גמישות משופרת של Managed Service for Apache Spark

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

המלצה: כדאי להשתמש במצב גמישות משופר של Managed Service for Apache Spark, שבו נתוני ערבוב ביניים לא מאוחסנים בעובדים משניים, כדי שניתן יהיה להקדים את העובדים המשניים או להקטין את קנה המידה שלהם בצורה בטוחה.

הגדרת חלוקה למחיצות וערבוב

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

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

הגדרת מחיצות

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

  • ‫spark.sql.files.maxPartitionBytes: הגודל המקסימלי של המחיצות כשקוראים נתונים מ-Cloud Storage. ברירת המחדל היא 128MB, שזה גודל מספיק לרוב האפליקציות שמעבדות פחות מ-100TB.

  • ‫spark.sql.shuffle.partitions: מספר המחיצות אחרי ביצוע ערבוב. ברירת המחדל היא 1000 עבור אשכולות של גרסאות תמונות מגרסה 2.2 ואילך. המלצה: מגדירים את הערך הזה כ-3 פעמים מספר יחידות ה-vCPU באשכול.

  • ‫spark.default.parallelism: מספר המחיצות שמוחזרות אחרי ביצוע טרנספורמציות של RDD שדורשות ערבוב, כמו join,‏ reduceByKey ו-parallelize. ברירת המחדל היא המספר הכולל של המעבדים הווירטואליים באשכול. כשמשתמשים ב-RDD במשימות Spark, אפשר להגדיר את המספר הזה ל-3x vCPU

הגבלת מספר הקבצים

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

הגדרת ביצוע שאילתות אדפטיבי (Spark 3)

הפעלה דינמית של שאילתות (מופעלת כברירת מחדל בגרסת תמונה 2.0 של Managed Service for Apache Spark) מספקת שיפורים בביצועים של משימות Spark, כולל:

הגדרות ברירת המחדל מתאימות לרוב תרחישי השימוש, אבל יכול להיות שיהיה לכם יתרון אם תגדירו את spark.sql.adaptive.advisoryPartitionSizeInBytes ל-spark.sql.files.maxPartitionBytes (ברירת המחדל היא 128MB).

אל תערבבו את הקלפים שלא לצורך

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

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

df.repartition("col_name").write().partitionBy("col_name").save("gs://...")

אחסון נתונים בפורמט Parquet או Avro

כברירת מחדל, Spark SQL קורא וכותב נתונים בקובצי Parquet דחוסים ב-Snappy. ‫Parquet הוא פורמט יעיל של קובץ עמודות שמאפשר ל-Spark לקרוא רק את הנתונים שהוא צריך כדי להריץ אפליקציה. זהו יתרון חשוב כשעובדים עם מערכי נתונים גדולים. גם פורמטים אחרים של עמודות, כמו Apache ORC, מניבים ביצועים טובים.

לנתונים לא עמודתיים,‏ Apache Avro מספק פורמט קובץ בינארי יעיל של שורות. למרות שבדרך כלל הוא איטי יותר מ-Parquet, הביצועים של Avro טובים יותר מאלה של פורמטים מבוססי-טקסט, כמו CSV או JSON.

אופטימיזציה של גודל הדיסק

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

כדי לעקוב אחרי קצב העברת הנתונים של הדיסק של העובד במסוףGoogle Cloud :

  1. לוחצים על שם האשכול בדף Clusters.
  2. לוחצים על הכרטיסייה VM INSTANCES (מופעי מכונות וירטואליות).
  3. לוחצים על השם של עובד.
  4. לוחצים על הכרטיסייה MONITORING (מעקב) וגוללים אל Disk Throughput (קצב העברת נתונים בדיסק) כדי לראות את קצב העברת הנתונים של העובד.

שיקולים לגבי הדיסק

אשכולות של Managed Service for Apache Spark שהם זמניים ולא נהנים מאחסון מתמיד יכולים להשתמש בכונני SSD מקומיים. אחסוני SSD מקומיים מחוברים פיזית לאשכול ומספקים תפוקה גבוהה יותר מדיסקים מתמידים (ראו את טבלת הביצועים). כונני SSD מקומיים זמינים בגודל קבוע של 375 גיגה-בייט, אבל אפשר להוסיף כמה כונני SSD כדי לשפר את הביצועים.

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

צירוף יחידות GPU לאשכול

‫Spark 3 תומך במעבדי GPU. אפשר להשתמש ב-GPU עם פעולת האתחול של RAPIDS כדי להאיץ את עבודות Spark באמצעות RAPIDS SQL Accelerator. פעולת האתחול של מנהל ההתקן של ה-GPU כדי להגדיר אשכול עם מעבדי GPU.

כשלים נפוצים במשימות ופתרונות

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

אין זיכרון פנוי

דוגמאות:

  • ‫Lost executor (מבצע שאבד)
  • ‪"java.lang.OutOfMemoryError: GC overhead limit exceeded"
  • ‫Container killed by YARN for exceeding memory limits

פתרונות אפשריים:

ערבוב של כשלים באחזור

דוגמאות:

  • ‫"FetchFailedException" (שגיאת Spark)
  • "החיבור אל... נכשל" (שגיאת Spark)
  • ‫"Failed to fetch" (שגיאת MapReduce)

בדרך כלל, הבעיה נגרמת בגלל הסרה מוקדמת של תהליכי worker שעדיין צריכים להעביר נתוני ערבוב.

סיבות אפשריות ופתרונות:

  • מכונות וירטואליות של worker שניתנות להפסקת פעולה נלקחו בחזרה, או שמכונות וירטואליות של worker שלא ניתן להפסיק את פעולתן הוסרו על ידי הכלי להתאמת קנה מידה אוטומטית. פתרון: משתמשים במצב גמישות משופר כדי לאפשר את ההפסקות או את ההרחבות של העובדים המשניים בצורה בטוחה.
  • ה-Executor או ה-Mapper קרסו בגלל שגיאת OutOfMemory. פתרון: הגדלת הזיכרון של executor או mapper.
  • יכול להיות שהשירות של Spark shuffle עמוס מדי. פתרון: צריך להקטין את מספר המחיצות של העבודה.

צמתים של YARN במצב UNHEALTHY

דוגמאות (מיומני YARN): 

...reported UNHEALTHY with details: 1/1 local-dirs usable space is below
configured utilization percentage/no more usable space
[ /hadoop/yarn/nm-local-dir : used space above threshold of 90.0% ]

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

  • פותחים את הדף Clusters של הפרויקט במסוף Google Cloud ולוחצים על שם האשכול.
  • לוחצים על הצגת יומנים.
  • סינון היומנים לפי hadoop-yarn-nodemanager.
  • מחפשים את המילה UNHEALTHY.

פתרונות אפשריים:

  • הזיכרון מטמון של המשתמש מאוחסן בספרייה שצוינה במאפיין yarn.nodemanager.local-dirs בקובץ yarn-site.xml file. הקובץ הזה נמצא במיקום /etc/hadoop/conf/yarn-site.xml. אפשר לבדוק את השטח הפנוי בנתיב /hadoop/yarn/nm-local-dir ולפנות מקום על ידי מחיקת תיקיית המטמון של המשתמש /hadoop/yarn/nm-local-dir/usercache.
  • אם ביומן מופיע הסטטוס UNHEALTHY, צריך ליצור מחדש את האשכול עם נפח דיסק גדול יותר, וכך להגדיל את המגבלה על קצב העברת הנתונים.

המטמון של המשתמש תופס מקום בדיסק

ב-Managed Service for Apache Spark, מטמון המשתמשים מאוחסן בספרייה שצוינה במאפיין yarn.nodemanager.local-dirs ב-yarn-site.xml (/etc/hadoop/conf/yarn-site.xml). ברוב המקרים, המיקום הוא /hadoop/yarn/nm-local-dir.

פתרונות אפשריים:

  • ניקוי ידני: בודקים את הנפח הפנוי בנתיב /hadoop/yarn/nm-local-dir, ואז מפנים מקום על ידי מחיקת תיקיית המטמון של המשתמש /hadoop/yarn/nm-local-dir/usercache.

  • פתרון לטווח הארוך: כדי לנהל את תהליך הניקוי, מגדירים את מאפייני האשכול הבאים:

    • yarn.nodemanager.localizer.cache.cleanup.interval-ms
    • yarn.nodemanager.localizer.cache.target-size-mb

    אם לא רוצים ליצור מחדש את האשכול, אפשר לעדכן את המאפיינים האלה ב-yarn-site.xml בכל הצמתים ואז להפעיל מחדש את NodeManager על ידי הפעלת הפקודה הבאה בצמתים של העובדים:

    sudo systemctl restart hadoop-yarn-nodemanager.service

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

העבודה נכשלת בגלל זיכרון לא מספיק של מנהל ההתקן

כשמריצים משימות במצב אשכול, המשימה נכשלת אם גודל הזיכרון של הזיכרון של צומת העובד.

דוגמה מיומני נהגים: 

'Exception in thread "main" java.lang.IllegalArgumentException:
Required AM memory (32768+3276 MB) is above the max threshold (12288 MB) of this cluster!
Please check the values of 'yarn.scheduler.maximum -allocation-mb' and/or 'yarn.nodemanager.resource.memory-mb'.'

פתרונות אפשריים:

  • מגדירים את spark:spark.driver.memory להיות קטן מ-yarn:yarn.scheduler.maximum-allocation-mb.
  • משתמשים באותו סוג מכונה לצומתי ה-master ולצומתי ה-worker.

המאמרים הבאים