1. סקירה כללית
ב-Lab הזה נסביר איך להגדיר ולהשתמש ב-Apache Spark וב-Jupyter notebooks ב-Managed Apache Spark.
מחברות Jupyter נמצאות בשימוש נרחב לניתוח נתונים לצורך גילוי תובנות ולבניית מודלים של למידת מכונה, כי הן מאפשרות להריץ את הקוד באופן אינטראקטיבי ולראות את התוצאות באופן מיידי.
עם זאת, ההגדרה והשימוש ב-Apache Spark וב-Jupyter Notebooks יכולים להיות מסובכים.
שירות מנוהל ל-Apache Spark מאפשר ליצור אשכול מנוהל של Apache Spark עם Apache Spark, Jupyter component ו-Component Gateway תוך כ-90 שניות, כך שהתהליך מהיר וקל.
מה תלמדו
ב-codelab הזה תלמדו איך:
- יצירת קטגוריה של Google Cloud Storage עבור האשכול
- יצירת אשכול מנוהל של Apache Spark עם Jupyter ו-Component Gateway,
- גישה לממשק המשתמש של JupyterLab באינטרנט ב-Managed Apache Spark
- יצירת מחברת באמצעות המחבר של Spark BigQuery Storage
- הפעלת משימת Spark ושרטוט התוצאות.
העלות הכוללת להרצת שיעור ה-Lab הזה ב-Google Cloud היא בערך 1$. פרטים מלאים על התמחור של Managed Apache Spark זמינים כאן.
2. יצירת פרויקט
נכנסים אל Google Cloud Platform Console בכתובת console.cloud.google.com ויוצרים פרויקט חדש:
בשלב הבא, כדי להשתמש במשאבים של Google Cloud, צריך להפעיל את החיוב במסוף Cloud.
העלות של התרגול ב-Codelab הזה לא אמורה להיות גבוהה מכמה דולרים, אבל היא יכולה להיות גבוהה יותר אם תחליטו להשתמש ביותר משאבים או אם תשאירו אותם פועלים. בחלק האחרון של ה-Codelab הזה מוסבר איך לנקות את הפרויקט.
משתמשים חדשים ב-Google Cloud Platform זכאים לתקופת ניסיון בחינם בשווי 300$.
3. הגדרת הסביבה
קודם כל, פותחים את Cloud Shell בלחיצה על הלחצן בפינה השמאלית העליונה של מסוף הענן:
אחרי ש-Cloud Shell נטען, מריצים את הפקודה הבאה כדי להגדיר את מזהה הפרויקט מהשלב הקודם**:**
gcloud config set project <project_id>
אפשר גם ללחוץ על הפרויקט בפינה הימנית העליונה של מסוף הענן כדי לראות את מזהה הפרויקט:
לאחר מכן, מפעילים את ממשקי ה-API של Managed Apache Spark, Compute Engine ו-BigQuery Storage.
gcloud services enable dataproc.googleapis.com \
compute.googleapis.com \
storage-component.googleapis.com \
bigquery.googleapis.com \
bigquerystorage.googleapis.com
אפשר גם לעשות את זה ב-Cloud Console. לוחצים על סמל התפריט בפינה השמאלית העליונה.
בתפריט הנפתח, בוחרים באפשרות 'API Manager'.
לוחצים על Enable APIs and Services.
מחפשים את ממשקי ה-API הבאים ומפעילים אותם:
- Compute Engine API
- Managed Apache Spark API
- BigQuery API
- BigQuery Storage API
4. יצירת קטגוריה ב-GCS
יוצרים קטגוריה של Google Cloud Storage באזור הכי קרוב לנתונים שלכם ונותנים לה שם ייחודי.
הוא ישמש לאשכול Apache Spark המנוהל.
REGION=us-central1
BUCKET_NAME=<your-bucket-name>
gsutil mb -c standard -l ${REGION} gs://${BUCKET_NAME}
הפלט הבא אמור להתקבל
Creating gs://<your-bucket-name>/...
5. יצירת אשכול מנוהל של Apache Spark באמצעות Jupyter ו-Component Gateway
יצירת האשכול
הגדרת משתני הסביבה של האשכול
REGION=us-central1
ZONE=us-central1-a
CLUSTER_NAME=spark-jupyter
BUCKET_NAME=<your-bucket-name>
לאחר מכן מריצים את פקודת gcloud הבאה כדי ליצור את האשכול עם כל הרכיבים הנדרשים לעבודה עם Jupyter באשכול.
gcloud beta dataproc clusters create ${CLUSTER_NAME} \
--region=${REGION} \
--image-version=2.2 \
--master-machine-type=n1-standard-4 \
--worker-machine-type=n1-standard-4 \
--bucket=${BUCKET_NAME} \
--optional-components=JUPYTER \
--enable-component-gateway
בזמן יצירת האשכול, אמור להופיע הפלט הבא
Waiting on operation [projects/spark-jupyter/regions/us-central1/operations/abcd123456].
Waiting for cluster creation operation...
יצירת האשכול תימשך כ-90 שניות, וכשהיא תהיה מוכנה תוכלו לגשת לאשכול מממשק המשתמש של מסוף Cloud של Managed Apache Spark.
בזמן ההמתנה, אפשר להמשיך לקרוא בהמשך כדי לקבל מידע נוסף על הדגלים שמשמשים בפקודת gcloud.
אחרי יצירת האשכול, הפלט הבא אמור להתקבל:
Created [https://dataproc.googleapis.com/v1beta2/projects/project-id/regions/us-central1/clusters/spark-jupyter] Cluster placed in zone [us-central1-a].
הדגלים שמשמשים בפקודה gcloud dataproc create
פירוט של האפשרויות שבהן נעשה שימוש בפקודה gcloud dataproc create
--region=${REGION}
מציינים את האזור והתחום שבהם ייווצר האשכול. כאן אפשר לראות את רשימת האזורים שבהם המינוי זמין.
--image-version=1.4
גרסת התמונה שבה רוצים להשתמש באשכול. כאן אפשר לראות את רשימת הגרסאות הזמינות.
--bucket=${BUCKET_NAME}
מציינים את קטגוריית Google Cloud Storage שיצרתם קודם לשימוש באשכול. אם לא תספקו קטגוריית GCS, היא תיווצר בשבילכם.
כאן גם יישמרו מחברות ה-Notebook שלכם, גם אם תמחקו את האשכול, כי מאגר ה-GCS לא יימחק.
--master-machine-type=n1-standard-4
--worker-machine-type=n1-standard-4
סוגי המכונות שבהן יש להשתמש באשכול Apache Spark המנוהל. כאן אפשר לראות רשימה של סוגי מכונות זמינים.
כברירת מחדל, נוצר צומת ראשי אחד ו-2 צמתים של עובדים אם לא מגדירים את הדגל –num-workers
--optional-components=ANACONDA,JUPYTER
הגדרת הערכים האלה עבור רכיבים אופציונליים תגרום להתקנה של כל הספריות הנדרשות ל-Jupyter ול-Anaconda (שנדרשת למחברות Jupyter) באשכול.
--enable-component-gateway
הפעלת Component Gateway יוצרת קישור ל-App Engine באמצעות Apache Knox ו-Inverting Proxy, שמאפשר גישה קלה, מאובטחת ומאומתת לממשקי האינטרנט של Jupyter ו-JupyterLab. המשמעות היא שכבר לא צריך ליצור מנהרות SSH.
הוא גם ייצור קישורים לכלים אחרים באשכול, כולל Yarn Resource manager ו-Spark History Server, שימושיים כדי לראות את הביצועים של העבודות ואת דפוסי השימוש באשכול.
6. יצירת מחברת Apache Spark
גישה לממשק האינטרנט של JupyterLab
אחרי שהאשכול מוכן, אפשר למצוא את הקישור של Component Gateway לממשק האינטרנט של JupyterLab. לשם כך, עוברים אל Managed Apache Spark Clusters - Cloud console (אשכולות מנוהלים של Apache Spark – מסוף Cloud), לוחצים על האשכול שיצרתם ועוברים לכרטיסייה Web Interfaces (ממשקי אינטרנט).
תראו שיש לכם גישה ל-Jupyter, שהוא ממשק ה-notebook הקלאסי, או ל-JupyterLab, שמוגדר כממשק המשתמש מהדור הבא של פרויקט Jupyter.
יש הרבה תכונות חדשות ונהדרות בממשק המשתמש של JupyterLab, ולכן אם אתם חדשים בשימוש במסמכי notebook או מחפשים את השיפורים האחרונים, מומלץ להשתמש ב-JupyterLab. לפי המסמכים הרשמיים, בסופו של דבר הוא יחליף את הממשק הקלאסי של Jupyter.
יצירת מחברת עם ליבת Python 3
בכרטיסייה של מרכז האפליקציות, לוחצים על סמל המחברת של Python 3 כדי ליצור מחברת עם ליבת Python 3 (לא ליבת PySpark). כך אפשר להגדיר את SparkSession במחברת ולכלול את spark-bigquery-connector שנדרש לשימוש ב-BigQuery Storage API.
שינוי השם של ה-Notebook
לוחצים לחיצה ימנית על שם המחברת בסרגל הצד בצד ימין או בסרגל הניווט העליון ומשנים את שם המחברת ל-BigQuery Storage & Spark DataFrames.ipynb.
הרצת קוד Spark ב-notebook
במחברת הזו תשתמשו ב-spark-bigquery-connector, כלי לקריאה ולכתיבה של נתונים בין BigQuery ל-Spark באמצעות BigQuery Storage API.
BigQuery Storage API מביא שיפורים משמעותיים לגישה לנתונים ב-BigQuery באמצעות פרוטוקול מבוסס-RPC. הוא תומך בקריאה ובכתיבה של נתונים במקביל, וגם בפורמטים שונים של סריאליזציה כמו Apache Avro ו-Apache Arrow. ברמה גבוהה, המשמעות היא שיפור משמעותי בביצועים, במיוחד כשמדובר במערכי נתונים גדולים.
בתא הראשון, בודקים את גרסת Scala של האשכול כדי לכלול את הגרסה הנכונה של קובץ ה-JAR של spark-bigquery-connector.
קלט [1]:
!scala -version
פלט [1]:
יוצרים סשן Spark וכוללים את חבילת spark-bigquery-connector.
אם גרסת Scala שלכם היא 2.11, צריך להשתמש בחבילה הבאה.
com.google.cloud.spark:spark-bigquery-with-dependencies_2.11:0.15.1-beta
אם גרסת Scala שלכם היא 2.12, צריך להשתמש בחבילה הבאה.
com.google.cloud.spark:spark-bigquery-with-dependencies_2.12:0.15.1-beta
קלט [2]:
from pyspark.sql import SparkSession
spark = SparkSession.builder \
.appName('BigQuery Storage & Spark DataFrames') \
.config('spark.jars.packages', 'com.google.cloud.spark:spark-bigquery-with-dependencies_2.11:0.15.1-beta') \
.getOrCreate()
הפעלת repl.eagerEval
הפעולה הזו תציג את התוצאות של DataFrames בכל שלב בלי הצורך להציג df.show() וגם תשפר את העיצוב של הפלט.
קלט [3]:
spark.conf.set("spark.sql.repl.eagerEval.enabled",True)
קריאת טבלה של BigQuery לתוך Spark DataFrame
יוצרים Spark DataFrame על ידי קריאת נתונים ממערך נתונים ציבורי של BigQuery. התהליך הזה משתמש ב- spark-bigquery-connector וב-BigQuery Storage API כדי לטעון את הנתונים אל אשכול Spark.
יוצרים Spark DataFrame וטוענים נתונים מקבוצת הנתונים הציבורית של BigQuery לגבי צפיות בדפים בוויקיפדיה. שימו לב שלא מריצים שאילתה על הנתונים כי משתמשים במחבר spark-bigquery כדי לטעון את הנתונים ל-Spark, שם יתבצע עיבוד הנתונים. כשמריצים את הקוד הזה, הטבלה לא נטענת בפועל כי מדובר בהערכה עצלה ב-Spark, והביצוע יתבצע בשלב הבא.
קלט [4]:
table = "bigquery-public-data.wikipedia.pageviews_2020"
df_wiki_pageviews = spark.read \
.format("bigquery") \
.option("table", table) \
.option("filter", "datehour >= '2020-03-01' AND datehour < '2020-03-02'") \
.load()
df_wiki_pageviews.printSchema()
פלט [4]:
בוחרים את העמודות הנדרשות ומחילים מסנן באמצעות where(), שהוא כינוי ל-filter().
כשמריצים את הקוד הזה, מופעלת פעולת Spark והנתונים נקראים מ-BigQuery Storage בשלב הזה.
קלט [5]:
df_wiki_en = df_wiki_pageviews \
.select("datehour", "wiki", "views") \
.where("views > 1000 AND wiki in ('en', 'en.m')") \
df_wiki_en
פלט [5]:
כדי לראות את הדפים המובילים, מקבצים לפי כותרת וממיינים לפי צפיות בדף
קלט [6]:
import pyspark.sql.functions as F
df_datehour_totals = df_wiki_en \
.groupBy("datehour") \
.agg(F.sum('views').alias('total_views'))
df_datehour_totals.orderBy('total_views', ascending=False)
פלט [6]:
7. שימוש בספריות של Python ליצירת תרשימים ב-notebook
אתם יכולים להשתמש בספריות שונות של תרשימים שזמינות ב-Python כדי ליצור תרשים של הפלט של משימות Spark.
המרת Spark DataFrame ל-Pandas DataFrame
ממירים את ה-Spark DataFrame ל-Pandas DataFrame ומגדירים את התאריך והשעה כאינדקס. זה שימושי אם רוצים לעבוד עם הנתונים ישירות ב-Python ולשרטט את הנתונים באמצעות ספריות רבות של שרטוט ב-Python.
קלט [7]:
spark.conf.set("spark.sql.execution.arrow.enabled", "true")
pandas_datehour_totals = df_datehour_totals.toPandas()
pandas_datehour_totals.set_index('datehour', inplace=True)
pandas_datehour_totals.head()
פלט [7]:
Plotting Pandas Dataframe
מייבאים את ספריית matplotlib שנדרשת להצגת התרשימים במחברת
קלט [8]:
import matplotlib.pyplot as plt
משתמשים בפונקציית התרשים של Pandas כדי ליצור תרשים קו מ-Pandas DataFrame.
קלט [9]:
pandas_datehour_totals.plot(kind='line',figsize=(12,6));
פלט [9]:
בדיקה שה-notebook נשמר ב-GCS
עכשיו אמור להיות לכם מסמך Jupyter notebook ראשון שפועל באשכול Apache Spark המנוהל. נותנים שם למחברת, והיא תינשמר אוטומטית בקטגוריית GCS שבה השתמשתם כשיצרתם את האשכול.
אפשר לבדוק את זה באמצעות פקודת gsutil הבאה ב-Cloud Shell
BUCKET_NAME=<your-bucket-name>
gsutil ls gs://${BUCKET_NAME}/notebooks/jupyter
הפלט הבא אמור להתקבל
gs://bucket-name/notebooks/jupyter/
gs://bucket-name/notebooks/jupyter/BigQuery Storage & Spark DataFrames.ipynb
8. טיפ לאופטימיזציה – שמירת נתונים במטמון בזיכרון
יכול להיות שיהיו תרחישים שבהם תרצו שהנתונים יהיו בזיכרון במקום לקרוא אותם מ-BigQuery Storage בכל פעם.
העבודה הזו תקרא את הנתונים מ-BigQuery ותדחוף את המסנן ל-BigQuery. לאחר מכן, הצבירה תחושב ב-Apache Spark.
import pyspark.sql.functions as F
table = "bigquery-public-data.wikipedia.pageviews_2020"
df_wiki_pageviews = spark.read \
.format("bigquery") \
.option("table", table) \
.option("filter", "datehour >= '2020-03-01' AND datehour < '2020-03-02'") \
.load()
df_wiki_en = df_wiki_pageviews \
.select("title", "wiki", "views") \
.where("views > 10 AND wiki in ('en', 'en.m')")
df_wiki_en_totals = df_wiki_en \
.groupBy("title") \
.agg(F.sum('views').alias('total_views'))
df_wiki_en_totals.orderBy('total_views', ascending=False)
אפשר לשנות את העבודה שלמעלה כך שתכלול מטמון של הטבלה, ועכשיו המסנן בעמודה של הוויקי יוחל בזיכרון על ידי Apache Spark.
import pyspark.sql.functions as F
table = "bigquery-public-data.wikipedia.pageviews_2020"
df_wiki_pageviews = spark.read \
.format("bigquery") \
.option("table", table) \
.option("filter", "datehour >= '2020-03-01' AND datehour < '2020-03-02'") \
.load()
df_wiki_all = df_wiki_pageviews \
.select("title", "wiki", "views") \
.where("views > 10")
# cache the data in memory
df_wiki_all.cache()
df_wiki_en = df_wiki_all \
.where("wiki in ('en', 'en.m')")
df_wiki_en_totals = df_wiki_en \
.groupBy("title") \
.agg(F.sum('views').alias('total_views'))
df_wiki_en_totals.orderBy('total_views', ascending=False)
אחר כך תוכלו לסנן לפי שפה אחרת של ויקיפדיה באמצעות הנתונים שבמטמון, במקום לקרוא שוב נתונים מאחסון BigQuery, ולכן הפעולה תהיה מהירה יותר.
df_wiki_de = df_wiki_all \
.where("wiki in ('de', 'de.m')")
df_wiki_de_totals = df_wiki_de \
.groupBy("title") \
.agg(F.sum('views').alias('total_views'))
df_wiki_de_totals.orderBy('total_views', ascending=False)
אפשר להסיר את המטמון באמצעות הפקודה
df_wiki_all.unpersist()
9. דוגמאות ל-notebook לתרחישי שימוש נוספים
ב-Managed Apache Spark GitHub repo יש מחברות Jupyter עם דפוסי Apache Spark נפוצים לטעינת נתונים, לשמירת נתונים ולשרטוט הנתונים באמצעות מגוון מוצרים של Google Cloud Platform וכלים בקוד פתוח:
10. הסרת המשאבים
כדי להימנע מחיובים מיותרים בחשבון GCP אחרי שתסיימו את המדריך למתחילים הזה:
- מוחקים את הקטגוריה של Cloud Storage עבור הסביבה שיצרתם
- מחיקת סביבת Managed Apache Spark.
אם יצרתם פרויקט רק בשביל ה-Codelab הזה, אתם יכולים גם למחוק אותו:
- במסוף GCP, נכנסים לדף Projects.
- ברשימת הפרויקטים, בוחרים את הפרויקט שרוצים למחוק ולוחצים על מחיקה.
- בתיבה, כותבים את מזהה הפרויקט ולוחצים על Shut down כדי למחוק את הפרויקט.
רישיון
עבודה זו מורשית תחת רישיון Creative Commons שמותנה בייחוס 3.0 כללי, ורישיון Apache 2.0.