טיפים לשיפור ביצועים של משימות 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

‫DataFrame API של Spark SQL הוא אופטימיזציה משמעותית של RDD API. אם אתם מבצעים אינטראקציה עם קוד שמשתמש ב-RDD, כדאי לקרוא את הנתונים כ-DataFrame לפני שמעבירים RDD בקוד. בקוד Java או Scala, כדאי להשתמש ב-Spark SQL Dataset API כקבוצת-על של 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. פעולות ערבוב הן יקרות, ולכן צריך להשתמש בהן בזהירות. הגדרה נכונה של התצורות של המחיצה אמורה להספיק כדי לאפשר ל-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 שלא ניתן להפסיק את הפעולה שלהן הוסרו על ידי קנה המידה האוטומטי. פתרון: משתמשים במצב גמישות משופר כדי לאפשר את ההפסקות או את ההרחבות של העובדים המשניים בצורה בטוחה.
  • ה-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.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.

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