טיפים לשיפור ביצועים של משימות 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.
  • לבצע אופטימיזציה של הגדרות החומרה והתוכנה של האשכול לכל משימה, ולשנות את הגדרות האשכול לפי הצורך.
  • כדאי לשדרג את גרסאות ה-image באשכולות חדשים כדי לקבל את תיקוני האבטחה, תיקוני הבאגים והאופטימיזציות העדכניים ביותר.
  • לפתור בעיות מהר יותר באשכול מבודד עם משימה אחת.
  • כדי לחסוך בעלויות, משלמים רק על זמן הריצה של אשכולות זמניים, ולא על זמן חוסר הפעילות בין משימות באשכול משותף.

שימוש ב-Spark SQL

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

שימוש ב-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.

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

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

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

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

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

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

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

‫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. ברירת המחדל היא המספר הכולל של המעבדים הווירטואליים באשכול. כשמשתמשים ב-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 במסוףCloud de Confiance :

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

שיקולים שקשורים לדיסק

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

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

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

‫Spark 3 תומך במעבדי GPU. להשתמש במעבדי GPU עם פעולת האתחול של CUDA-X כדי להאיץ משימות Spark באמצעות ספריית NVIDIA cuDF ל-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 שעדיין יש להם נתוני ערבוב להצגה.

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

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

צומתי YARN לא תקינים

דוגמאות (מיומני 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 בפרויקט במסוף Cloud de Confiance ולוחצים על שם האשכול.
  • לוחצים על 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.
  • משתמשים באותו סוג מכונה לצומת הראשי ולצומתי העובדים.

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