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

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

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

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

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

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

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

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

מצד שני, אשכולות זמניים מאפשרים לכם:

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

שימוש ב-Spark SQL

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

שימוש ב-Apache Spark 3

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

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

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

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

איך משתמשים בהתאמה אוטומטית לעומס (autoscaling) ב-Managed Service for Apache Spark

התאמה אוטומטית לעומס (autoscaling) ב-Managed Service for Apache Spark מוסיפה ומסירה באופן דינמי עובדים של Managed Service for Apache Spark מאשכול, כדי לוודא שלמשימות Spark יש את המשאבים הדרושים להשלמה מהירה.

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

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

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

המלצה: השתמשו ב-Managed Service for Apache Spark Enhanced Flexibility Mode, שלא מאחסן נתוני ביניים של ערבוב (shuffle) ב-workers משניים, כדי שיהיה אפשר להפסיק את הפעולה של ה-workers המשניים או לצמצם אותם אנכית בהתאם לעומס (scale down) באופן בטוח.

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

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

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

הגדרת מחיצות

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

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

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

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

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

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

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

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

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

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

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

הוספת יחידות GPU לאשכול

‫Spark 3 תומך במעבדי GPU. להשתמש במעבדים גרפיים עם פעולת האתחול של CUDA-X כדי להאיץ משימות Spark באמצעות ספריית NVIDIA cuDF for Apache Spark. פעולת האתחול של מנהל ההתקן של ה-GPU כדי להגדיר אשכול עם מעבדי GPU.

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

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

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

דוגמאות:

  • "Lost executor"
  • "java.lang.OutOfMemoryError: GC overhead limit exceeded"
  • "Container killed by YARN for exceeding memory limits"

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

כשלים באחזור של הפעלה אקראית

דוגמאות:

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

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

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

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

צומתי 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 ולוחצים על שם האשכול.
  • לוחצים על VIEW LOGS.
  • סינון היומנים לפי 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.
  • משתמשים באותו סוג מכונה לצומת הראשי ולצומתי העובדים.

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