1. סקירה כללית
בשיעור Codelab הזה נסביר איך ליצור פייפליין לעיבוד נתונים באמצעות Apache Spark עם Dataproc ב-Google Cloud Platform. תרחיש שימוש נפוץ במדעי הנתונים ובהנדסת נתונים הוא קריאת נתונים ממיקום אחסון אחד, ביצוע טרנספורמציות בנתונים וכתיבתם במיקום אחסון אחר. בין השינויים הנפוצים: שינוי התוכן של הנתונים, הסרת מידע מיותר ושינוי סוגי הקבצים.
בשיעור Codelab הזה נלמד על Apache Spark, נריץ צינור עיבוד נתונים לדוגמה באמצעות Dataproc עם PySpark (Python API של Apache Spark), BigQuery, Google Cloud Storage ונתונים מ-Reddit.
2. מבוא ל-Apache Spark (אופציונלי)
לפי האתר, Apache Spark הוא מנוע ניתוח מאוחד לעיבוד נתונים בקנה מידה גדול. היא מאפשרת לנתח ולעבד נתונים במקביל ובזיכרון, מה שמאפשר חישוב מקבילי נרחב בכמה מכונות וצמתים שונים. הוא הושק במקור בשנת 2014 כשדרוג ל-MapReduce המסורתי, והוא עדיין אחד מהמסגרות הפופולריות ביותר לביצוע חישובים בקנה מידה גדול. Apache Spark כתוב ב-Scala, ולכן יש לו ממשקי API ב-Scala, ב-Java, ב-Python וב-R. הוא מכיל שפע של ספריות כמו Spark SQL להרצת שאילתות SQL על הנתונים, Spark Streaming להזרמת נתונים, MLlib ללמידת מכונה ו-GraphX לעיבוד גרפים, וכולן פועלות במנוע Apache Spark.
Spark יכול לפעול לבד או להשתמש בשירות לניהול משאבים כמו Yarn, Mesos או Kubernetes לצורך שינוי גודל. בשיעור Codelab הזה נשתמש ב-Dataproc, שמבוסס על Yarn.
הנתונים ב-Spark נטענו במקור לזיכרון למה שנקרא RDD, או מערך נתונים מבוזר עמיד. מאז, הפיתוח של Spark כלל הוספה של שני סוגים חדשים של נתונים בסגנון עמודות: Dataset, שהוא מוקלד, ו-Dataframe, שהוא לא מוקלד. באופן כללי, RDDs מתאימים לכל סוג של נתונים, בעוד ש-Datasets ו-Dataframes מותאמים לנתונים טבלאיים. מערכי נתונים זמינים רק ב-Java וב-Scala API, ולכן נמשיך להשתמש ב-PySpark Dataframe API בשיעור Codelab הזה. מידע נוסף מופיע במאמרי העזרה בנושא Apache Spark.
3. תרחיש לדוגמה
מהנדסי נתונים צריכים לעיתים קרובות שהנתונים יהיו נגישים בקלות למדעני נתונים. עם זאת, לעיתים קרובות הנתונים לא נקיים בהתחלה (קשה להשתמש בהם לניתוח במצבם הנוכחי) וצריך לנקות אותם לפני שאפשר להפיק מהם תועלת. דוגמה לכך היא נתונים שגורדו מהאינטרנט, שעשויים להכיל קידודים מוזרים או תגי HTML מיותרים.
בשיעור Lab הזה תטענו קבוצת נתונים מ-BigQuery בצורה של פוסטים ב-Reddit לאשכול Spark שמתארח ב-Dataproc, תחלצו מידע שימושי ותשמרו את הנתונים המעובדים כקובצי CSV דחוסים ב-Google Cloud Storage.
החוקר הראשי של נתונים בחברה שלכם מעוניין שהצוותים שלו יעבדו על בעיות שונות של עיבוד שפה טבעית. הם רוצים לנתח את הנתונים בסאברדיט r/food. תצרו צינור לעיבוד נתונים של dump, שיתחיל במילוי חוסרים מ-2017 עד 2019.
4. גישה ל-BigQuery דרך BigQuery Storage API
שליפת נתונים מ-BigQuery באמצעות השיטה tabledata.list API עלולה להיות ארוכה ולא יעילה ככל שכמות הנתונים גדלה. השיטה הזו מחזירה רשימה של אובייקטים בפורמט JSON, ונדרשת קריאה רציפה של דף אחד בכל פעם כדי לקרוא מערך נתונים שלם.
BigQuery Storage API משפר באופן משמעותי את הגישה לנתונים ב-BigQuery באמצעות פרוטוקול מבוסס-RPC. הוא תומך בקריאה ובכתיבה של נתונים במקביל, וגם בפורמטים שונים של סריאליזציה כמו Apache Avro ו-Apache Arrow. ברמה גבוהה, המשמעות היא שיפור משמעותי בביצועים, במיוחד כשמדובר במערכי נתונים גדולים יותר.
בשיעור Codelab הזה נשתמש ב-מחבר spark-bigquery כדי לקרוא ולכתוב נתונים בין BigQuery ל-Spark.
5. יצירת פרויקט
נכנסים אל Google Cloud Platform Console בכתובת console.cloud.google.com ויוצרים פרויקט חדש:
בשלב הבא, כדי להשתמש במשאבים של Google Cloud, צריך להפעיל את החיוב במסוף Cloud.
העלות של ה-Codelab הזה לא אמורה להיות גבוהה מכמה דולרים, אבל היא יכולה להיות גבוהה יותר אם תחליטו להשתמש ביותר משאבים או אם תשאירו אותם פועלים. בקטע האחרון של שיעור ה-Codelab הזה תלמדו איך לנקות את הפרויקט.
משתמשים חדשים ב-Google Cloud Platform זכאים לתקופת ניסיון בחינם בשווי 300$.
6. הגדרת הסביבה
עכשיו תצטרכו להגדיר את הסביבה שלכם על ידי:
- הפעלת ממשקי ה-API של Compute Engine, Dataproc ו-BigQuery Storage
- הגדרת הגדרות הפרויקט
- יצירת אשכול Dataproc
- יצירת קטגוריה של Google Cloud Storage
הפעלת ממשקי API והגדרת הסביבה
פותחים את Cloud Shell בלחיצה על הלחצן בפינה השמאלית העליונה של Cloud Console.
אחרי ש-Cloud Shell נטען, מריצים את הפקודות הבאות כדי להפעיל את Compute Engine, Dataproc ו-BigQuery Storage APIs:
gcloud services enable compute.googleapis.com \
dataproc.googleapis.com \
bigquerystorage.googleapis.com
מגדירים את מזהה הפרויקט של הפרויקט. כדי למצוא אותו, עוברים לדף בחירת הפרויקט ומחפשים את הפרויקט. יכול להיות שהשם הזה שונה משם הפרויקט.
מריצים את הפקודה הבאה כדי להגדיר את מזהה הפרויקט:
gcloud config set project <project_id>
בוחרים אזור לפרויקט מהרשימה כאן. לדוגמה, us-central1.
gcloud config set dataproc/region <region>
בוחרים שם לאשכול Dataproc ויוצרים בשבילו משתנה סביבה.
CLUSTER_NAME=<cluster_name>
יצירת אשכול Dataproc
יוצרים אשכול Dataproc באמצעות הפקודה הבאה:
gcloud beta dataproc clusters create ${CLUSTER_NAME} \
--worker-machine-type n1-standard-8 \
--num-workers 8 \
--image-version 1.5-debian \
--initialization-actions gs://dataproc-initialization-actions/python/pip-install.sh \
--metadata 'PIP_PACKAGES=google-cloud-storage' \
--optional-components=ANACONDA \
--enable-component-gateway
הפקודה הזו תימשך כמה דקות. פירוט הפקודה:
הפעולה הזו תתחיל את היצירה של אשכול Dataproc עם השם שציינתם קודם. שימוש ב-beta API יאפשר לכם להשתמש בתכונות בטא של Dataproc, כמו Component Gateway.
gcloud beta dataproc clusters create ${CLUSTER_NAME}
ההגדרה הזו תגדיר את סוג המכונה שבה ישתמשו העובדים.
--worker-machine-type n1-standard-8
המספר הזה יגדיר את מספר העובדים שיהיו באשכול.
--num-workers 8
הפעולה הזו תגדיר את גרסת האימג' של Dataproc.
--image-version 1.5-debian
הפעולה הזו תגדיר את פעולות האתחול לשימוש באשכול. כאן אתם כוללים את פעולת האתחול של pip.
--initialization-actions gs://dataproc-initialization-actions/python/pip-install.sh
אלה המטא-נתונים שצריך לכלול באשכול. כאן אתם מספקים מטא-נתונים לפעולת האתחול pip.
--metadata 'PIP_PACKAGES=google-cloud-storage'
הפעולה הזו תגדיר את המרכיבים האופציונליים להתקנה באשכול.
--optional-components=ANACONDA
הפעולה הזו תפעיל את Component Gateway, שיאפשר לכם להשתמש ב-Component Gateway של Dataproc כדי להציג ממשקי משתמש נפוצים כמו Zeppelin, Jupyter או Spark History.
--enable-component-gateway
למידע נוסף על Dataproc, אפשר לעיין ב-codelab הזה.
יצירת קטגוריה של Google Cloud Storage
תצטרכו קטגוריה של Google Cloud Storage בשביל פלט העבודה. קובעים שם ייחודי לקטגוריה ומריצים את הפקודה הבאה כדי ליצור קטגוריה חדשה. השמות של הקטגוריות הם ייחודיים לכל הפרויקטים ב-Google Cloud עבור כל המשתמשים, ולכן יכול להיות שתצטרכו לנסות כמה פעמים עם שמות שונים. אם לא מקבלים את השגיאה ServiceException, סימן שהקטגוריה נוצרה בהצלחה.
BUCKET_NAME=<bucket_name>
gsutil mb gs://${BUCKET_NAME}
7. ניתוח נתונים לצורך גילוי תובנות
לפני שמבצעים עיבוד מקדים, כדאי לקרוא מידע נוסף על אופי הנתונים שבהם עוסקים. כדי לעשות את זה, תצטרכו לבחון שתי שיטות לניתוח נתונים. קודם תצפו בנתונים גולמיים באמצעות ממשק האינטרנט של BigQuery, ואז תחשבו את מספר הפוסטים בכל קהילת Reddit באמצעות PySpark ו-Dataproc.
שימוש בממשק המשתמש של BigQuery באינטרנט
כדאי להתחיל להשתמש בממשק המשתמש של BigQuery באינטרנט כדי לצפות בנתונים. בסמל התפריט ב-Cloud Console, גוללים למטה ולוחצים על BigQuery כדי לפתוח את ממשק האינטרנט של BigQuery.
אחר כך מריצים את הפקודה הבאה בעורך השאילתות בממשק האינטרנט של BigQuery. הפונקציה תחזיר 10 שורות מלאות של הנתונים מינואר 2017:
select * from fh-bigquery.reddit_posts.2017_01 limit 10;
אפשר לגלול לאורך הדף כדי לראות את כל העמודות הזמינות וגם כמה דוגמאות. בפרט, תוכלו לראות שתי עמודות שמייצגות את התוכן הטקסטואלי של כל פוסט: 'כותרת' ו-'selftext', שהיא גוף הפוסט. שימו לב גם לעמודות אחרות, כמו 'created_utc' שבה מופיע הזמן ב-UTC שבו נוצר הפוסט, ו-'subreddit' שבה מופיע הסאברדיט שבו הפוסט קיים.
ביצוע של משימת PySpark
מריצים את הפקודות הבאות ב-Cloud Shell כדי לשכפל את המאגר עם הקוד לדוגמה ולעבור לספרייה הנכונה:
cd
git clone https://github.com/GoogleCloudPlatform/cloud-dataproc
אפשר להשתמש ב-PySpark כדי לקבוע כמה פוסטים יש בכל subreddit. אפשר לפתוח את Cloud Editor ולקרוא את הסקריפט cloud-dataproc/codelabs/spark-bigquery לפני שמריצים אותו בשלב הבא:
לוחצים על הכפתור 'Open Terminal' (פתיחת הטרמינל) ב-Cloud Editor כדי לחזור ל-Cloud Shell ולהריץ את הפקודה הבאה להפעלת עבודת PySpark הראשונה:
cd ~/cloud-dataproc/codelabs/spark-bigquery
gcloud dataproc jobs submit pyspark --cluster ${CLUSTER_NAME} \
--jars gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar \
--driver-log-levels root=FATAL \
counts_by_subreddit.py
הפקודה הזו מאפשרת לשלוח משימות ל-Dataproc דרך Jobs API. כאן מציינים את סוג העבודה כpyspark. אפשר לספק את שם האשכול, פרמטרים אופציונליים ואת שם הקובץ שמכיל את העבודה. כאן מציינים את הפרמטר --jars שמאפשר לכלול את spark-bigquery-connector במשרה. אפשר גם להגדיר את רמות הפלט של היומן באמצעות --driver-log-levels root=FATAL, וכך למנוע את כל פלט היומן חוץ משגיאות. היומנים של Spark נוטים להיות רועשים למדי.
הפעולה תימשך כמה דקות, והפלט הסופי אמור להיראות כך:
8. הכרת ממשקי המשתמש של Dataproc ו-Spark
כשמריצים משימות Spark ב-Dataproc, יש לכם גישה לשני ממשקי משתמש לבדיקת הסטטוס של המשימות או האשכולות. האפשרות הראשונה היא ממשק המשתמש של Dataproc. כדי להגיע אליו, לוחצים על סמל התפריט וגוללים למטה אל Dataproc. כאן אפשר לראות את הזיכרון הפנוי הנוכחי, את הזיכרון בהמתנה ואת מספר העובדים.
אפשר גם ללחוץ על הכרטיסייה 'משימות' כדי לראות את המשימות שהושלמו. כדי לראות פרטים על משרה מסוימת, כמו היומנים והפלט של המשרות האלה, לוחצים על מזהה המשרה. 
אפשר גם להציג את ממשק המשתמש של Spark. בדף המשימה, לוחצים על חץ החזרה ואז על 'ממשקי אינטרנט'. אמורות להופיע כמה אפשרויות בקטע 'שער רכיבים'. אפשר להפעיל הרבה מהרכיבים האלה דרך רכיבים אופציונליים כשמגדירים את האשכול. במעבדה הזו, לוחצים על Spark History Server.
אמור להיפתח החלון הבא:
כל המשימות שהושלמו יופיעו כאן, ואפשר ללחוץ על כל application_id כדי לקבל מידע נוסף על המשימה. באופן דומה, אפשר ללחוץ על 'הצגת בקשות לא שלמות' בתחתית דף הנחיתה כדי לראות את כל המשרות שמוצגות כרגע.
9. הפעלת משימת מילוי חוסרים
עכשיו תפעילו משימה שתטען נתונים לזיכרון, תחלץ את המידע הדרוש ותעביר את הפלט לקטגוריה של Google Cloud Storage. תחלצו את הערכים של 'כותרת', 'גוף' (טקסט גולמי) ו'חותמת זמן של יצירה' לכל תגובה ב-Reddit. לאחר מכן, תמירו את הנתונים האלה לקובץ CSV, תדחסו אותו ותטענו אותו ל-bucket עם URI של gs://${BUCKET_NAME}/reddit_posts/YYYY/MM/food.csv.gz.
אפשר לעיין שוב ב-Cloud Editor כדי לקרוא את הקוד של cloud-dataproc/codelabs/spark-bigquery/backfill.sh, שהוא סקריפט wrapper להרצת הקוד ב-cloud-dataproc/codelabs/spark-bigquery/backfill.py.
cd ~/cloud-dataproc/codelabs/spark-bigquery
bash backfill.sh ${CLUSTER_NAME} ${BUCKET_NAME}
אחרי זמן קצר אמורות להופיע הרבה הודעות על השלמת עבודות. יכול להיות שיידרשו עד 15 דקות להשלמת העבודה. אפשר גם לבדוק את קטגוריית האחסון כדי לוודא שהנתונים יוצאו בהצלחה באמצעות gsutil. אחרי שכל העבודות מסתיימות, מריצים את הפקודה הבאה:
gsutil ls gs://${BUCKET_NAME}/reddit_posts/*/*/food.csv.gz
הפלט הבא אמור להתקבל:
השלמת בהצלחה מילוי חוסרים בנתוני התגובות שלך ב-Reddit. אם אתם רוצים לדעת איך לבנות מודלים על בסיס הנתונים האלה, אתם יכולים להמשיך אל ה-codelab של Spark-NLP.
10. הסרת המשאבים
כדי להימנע מחיובים מיותרים בחשבון GCP אחרי שתסיימו את המדריך למתחילים הזה:
- מוחקים את הקטגוריה של Cloud Storage עבור הסביבה שיצרתם
- מחיקת סביבת Dataproc.
אם יצרתם פרויקט רק בשביל ה-Codelab הזה, אתם יכולים גם למחוק אותו:
- במסוף GCP, נכנסים לדף Projects.
- ברשימת הפרויקטים, בוחרים את הפרויקט שרוצים למחוק ולוחצים על מחיקה.
- בתיבה, כותבים את מזהה הפרויקט ולוחצים על Shut down כדי למחוק את הפרויקט.
רישיון
עבודה זו מורשית תחת רישיון Creative Commons שמותנה בייחוס 3.0 כללי, ורישיון Apache 2.0.