עבודה עם פרוצדורות מאוחסנות ב-Apache Spark

אתם יכולים להשתמש בפרוצדורות מאוחסנות של Spark ב-BigQuery כדי להריץ עומסי עבודה מורכבים של עיבוד נתונים שכתובים ב-Python, ב-Java וב-Scala ישירות במחסן הנתונים. הפעלת הפרוצדורות האלה באמצעות שאילתות GoogleSQL רגילות מאפשרת לכם לשלב משימות מתקדמות של הנדסת נתונים ומדעי נתונים בצינורות עיבוד הנתונים הקיימים של SQL, בדומה להפעלת פרוצדורות מאוחסנות של SQL.

לפני שמתחילים

כדי ליצור פרוצדורה מאוחסנת ל-Spark, בקשו מהאדמין ליצור חיבור ל-Spark ולשתף אותו איתכם. האדמין צריך גם לתת לחשבון השירות שמשויך לחיבור את ההרשאות הנדרשות לניהול זהויות והרשאות גישה (IAM).

התפקידים הנדרשים

כדי לקבל את ההרשאות שדרושות לביצוע המשימות שמפורטות במאמר הזה, בקשו מהאדמין להקצות לכם את התפקידים הבאים ב-IAM:

• יצירת פרוצדורה מאוחסנת ל-Spark:
  • BigQuery Data Editor (roles/bigquery.dataEditor) במערך הנתונים שבו יוצרים את הפרוצדורה המאוחסנת
  • אדמין של חיבור ל-BigQuery (roles/bigquery.connectionAdmin) בחיבור שבו משתמש התהליך המאוחסן
  • BigQuery Job User (roles/bigquery.jobUser) בפרויקט
• קריאה לתהליך מאוחסן ב-Spark:
  • ‫BigQuery Metadata Viewer ‏ (roles/bigquery.metadataViewer) במערך הנתונים שבו מאוחסנת הפרוצדורה המאוחסנת
  • BigQuery Connection User (roles/bigquery.connectionUser) בחיבור
  • BigQuery Job User (roles/bigquery.jobUser) בפרויקט

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

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

ההרשאות הנדרשות

כדי לבצע את המשימות שבמסמך הזה, נדרשות ההרשאות הבאות:

• יוצרים חיבור:
  • bigquery.connections.create
  • bigquery.connections.list
• יוצרים תהליך מאוחסן ל-Spark:
  • bigquery.routines.create
  • bigquery.connections.delegate
  • bigquery.jobs.create
• קריאה לתהליך מאוחסן ב-Spark:
  • bigquery.routines.get
  • bigquery.connections.use
  • bigquery.jobs.create

יכול להיות שתקבלו את ההרשאות האלה באמצעות תפקידים בהתאמה אישית או תפקידים מוגדרים מראש אחרים.

שיקולים בקשר למיקום

צריך ליצור פרוצדורה מאוחסנת ל-Spark באותו מיקום שבו נמצא החיבור, כי הפרוצדורה המאוחסנת פועלת באותו מיקום שבו נמצא החיבור. לדוגמה, כדי ליצור פרוצדורה מאוחסנת במספר אזורים בארה"ב, צריך להשתמש בחיבור שנמצא במספר אזורים בארה"ב.

תמחור

• החיובים על הרצת פרוצדורות של Spark ב-BigQuery דומים לחיובים על הרצת פרוצדורות של Spark ב-Managed Service for Apache Spark. למידע נוסף על התמחור של Managed Service for Apache Spark

• אפשר להשתמש בפרוצדורות המאוחסנות של Spark עם מודל התמחור על פי דרישה ועם כל אחת ממהדורות BigQuery. החיוב על פרוצדורות Spark מתבצע באמצעות מודל התשלום לפי שימוש של מהדורת BigQuery Enterprise בכל המקרים, בלי קשר למודל התמחור של המחשוב שבו אתם משתמשים בפרויקט.

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

• למרות שהפרוצדורות המאוחסנות של Spark משתמשות במנוע ביצוע של Spark, לא תראו חיובים נפרדים על הביצוע של Spark. כמו שצוין, החיובים התואמים מדווחים כמק"ט של תשלום לפי שימוש במהדורת BigQuery Enterprise.

• קריאת נתונים מטבלאות ב-BigQuery באמצעות פרוצדורות מאוחסנות ל-Spark לא כרוכה בחיובים על Storage Read API. פעולות קריאה בסטרימינג כלולות בהרצה של הפרוצדורה המאוחסנת ללא תשלום נוסף. אם קוראים נתונים בין אזורים, חלים תעריפים רגילים של תעבורת נתונים יוצאת ברשת.

• אין תוכנית בחינם לתהליכים מאוחסנים ב-Spark.

יצירת פרוצדורה מאוחסנת ל-Spark

צריך ליצור את הפרוצדורה המאוחסנת באותו מיקום שבו נמצא החיבור שבו אתם משתמשים.

אם גוף התהליך המאוחסן גדול מ-1MB, מומלץ לשים את התהליך המאוחסן בקובץ בקטגוריה של Cloud Storage במקום להשתמש בקוד מוטבע. ב-BigQuery יש שתי שיטות ליצירת פרוצדורה מאוחסנת ל-Spark באמצעות Python:

• אם רוצים להשתמש בהצהרה CREATE PROCEDURE, צריך להשתמש בעורך השאילתות של SQL.
• אם אתם רוצים להקליד קוד Python ישירות, השתמשו בכלי העריכה של PySpark. אתם יכולים לשמור את הקוד כפרוצדורה מאוחסנת.

שימוש בעורך שאילתות SQL

כדי ליצור פרוצדורה מאוחסנת ל-Spark בעורך השאילתות ב-SQL:

1. נכנסים לדף BigQuery.

   כניסה ל-BigQuery

2. בעורך השאילתות, מוסיפים את הקוד לדוגמה של ההצהרה CREATE PROCEDURE שמופיעה.

   לחלופין, בחלונית Explorer לוחצים על החיבור בפרויקט שבו השתמשתם כדי ליצור את משאב החיבור. כדי ליצור פרוצדורה מאוחסנת ל-Spark, לוחצים על Create stored procedure.

   Python

   כדי ליצור פרוצדורות מאוחסנות ל-Spark ב-Python, משתמשים בקוד לדוגמה הבא:

   CREATE OR REPLACE PROCEDURE `PROJECT_ID`.DATASET.PROCEDURE_NAME(PROCEDURE_ARGUMENT)
    WITH CONNECTION `CONNECTION_PROJECT_ID.CONNECTION_REGION.CONNECTION_ID`
    OPTIONS (
        engine="SPARK", runtime_version="RUNTIME_VERSION",
        main_file_uri=["MAIN_PYTHON_FILE_URI"]);
    LANGUAGE PYTHON [AS PYSPARK_CODE]
   

   ‫Java או Scala

   כדי ליצור פרוצדורה מאוחסנת ל-Spark ב-Java או ב-Scala עם האפשרות main_file_uri, משתמשים בקוד לדוגמה הבא:

   CREATE [OR REPLACE] PROCEDURE `PROJECT_ID`.DATASET.PROCEDURE_NAME(PROCEDURE_ARGUMENT)
    WITH CONNECTION `CONNECTION_PROJECT_ID.CONNECTION_REGION.CONNECTION_ID`
    OPTIONS (
        engine="SPARK", runtime_version="RUNTIME_VERSION",
        main_file_uri=["MAIN_JAR_URI"]);
    LANGUAGE JAVA|SCALA
   

   כדי ליצור פרוצדורה מאוחסנת ל-Spark ב-Java או ב-Scala עם האפשרויות main_class ו-jar_uris, משתמשים בקוד לדוגמה הבא:

   CREATE [OR REPLACE] PROCEDURE `PROJECT_ID`.DATASET.PROCEDURE_NAME(PROCEDURE_ARGUMENT)
    WITH CONNECTION `CONNECTION_PROJECT_ID.CONNECTION_REGION.CONNECTION_ID`
    OPTIONS (
        engine="SPARK", runtime_version="RUNTIME_VERSION",
        main_class=["CLASS_NAME"],
        jar_uris=["URI"]);
    LANGUAGE JAVA|SCALA
   

   מחליפים את מה שכתוב בשדות הבאים:

   • ‫PROJECT_ID: הפרויקט שבו רוצים ליצור את הפרוצדורה המאוחסנת, לדוגמה myproject.
   • ‫DATASET: מערך הנתונים שבו רוצים ליצור את הפרוצדורה המאוחסנת, לדוגמה mydataset.
   • ‫PROCEDURE_NAME: השם של הפרוצדורה המאוחסנת שרוצים להריץ ב-BigQuery, לדוגמה mysparkprocedure.
   • ‫PROCEDURE_ARGUMENT: פרמטר להזנת הארגומנטים של הקלט.

     בפרמטר הזה, מציינים את השדות הבאים:

     • ‫ARGUMENT_MODE: המצב של הארגומנט.

       הערכים התקינים כוללים את IN, OUT ו-INOUT. ערך ברירת המחדל הוא IN.

     • ‫ARGUMENT_NAME: שם הארגומנט.
     • ‫ARGUMENT_TYPE: סוג הארגומנט.
       

     לדוגמה: myproject.mydataset.mysparkproc(num INT64).

     למידע נוסף, עיינו בקטע העברת ערך כפרמטר IN או בקטע הפרמטרים OUT ו-INOUT במאמר הזה.

   • ‫CONNECTION_PROJECT_ID: הפרויקט שמכיל את החיבור להרצת הפרוצדורה של Spark.
   • ‫CONNECTION_REGION: האזור שמכיל את החיבור להרצת פרוצדורת Spark, לדוגמה us.
   • ‫CONNECTION_ID: מזהה החיבור, לדוגמה myconnection.

     כשמציגים את פרטי החיבור במסוף Google Cloud , מזהה החיבור הוא הערך שמופיע בקטע האחרון של מזהה החיבור המלא שמוצג ב-Connection ID, לדוגמה projects/myproject/locations/connection_location/connections/myconnection.

   • ‫RUNTIME_VERSION: גרסת סביבת זמן הריצה של Spark, לדוגמה 2.2.
   • ‫MAIN_PYTHON_FILE_URI: הנתיב לקובץ PySpark, לדוגמה gs://mybucket/mypysparkmain.py.

     לחלופין, אם רוצים להוסיף את גוף הפרוצדורה המאוחסנת בהצהרה CREATE PROCEDURE, צריך להוסיף את PYSPARK_CODE אחרי LANGUAGE PYTHON AS, כמו שמוצג בדוגמה בקטע שימוש בקוד מוטבע במסמך הזה.

   • ‫PYSPARK_CODE: ההגדרה של אפליקציית PySpark בהצהרה CREATE PROCEDURE, אם רוצים להעביר את גוף התהליך בתוך השורה.

     הערך הוא מחרוזת מילולית. אם הקוד כולל מירכאות ולוכסנים הפוכים, צריך להוסיף להם תו בריחה או לייצג אותם כמחרוזת גולמית. לדוגמה, אפשר לייצג את קוד ההחזרה "\n"; באחת מהדרכים הבאות:

     • מחרוזת במירכאות: "return \"\\n\";". גם מירכאות וגם לוכסנים הפוכים מסומנים בתווי בריחה (escape).
     • מחרוזת עם שלושה סימני מירכאות: """return "\\n";""". לוכסנים הפוכים מתווספים כתווי בריחה (escape), אבל לא מירכאות.
     • מחרוזת גולמית: r"""return "\n";""". אין צורך בתווי בריחה (escape).
     כדי ללמוד איך להוסיף קוד PySpark בתוך השורה, קראו את המאמר שימוש בקוד בתוך השורה.
   • ‫MAIN_JAR_URI: הנתיב של קובץ ה-JAR שמכיל את המחלקה main, לדוגמה gs://mybucket/my_main.jar.
   • ‫CLASS_NAME: השם המלא של מחלקה בקובץ JAR שמוגדר עם האפשרות jar_uris, לדוגמה, com.example.wordcount.
   • ‫URI: הנתיב של קובץ ה-JAR שמכיל את המחלקה שצוינה במחלקה main, לדוגמה, gs://mybucket/mypysparkmain.jar.

   לאפשרויות נוספות שאפשר לציין ב-OPTIONS, ראו את רשימת האפשרויות של הפרוצדורה.

שימוש בכלי העריכה של PySpark

כשיוצרים פרוצדורה באמצעות עורך PySpark, לא צריך להשתמש בהצהרה CREATE PROCEDURE. במקום זאת, הוסיפו את קוד ה-Python ישירות בעורך PySpark ושמרו או הריצו את הקוד.

כדי ליצור פרוצדורה מאוחסנת ל-Spark בעורך PySpark, מבצעים את השלבים הבאים:

1. נכנסים לדף BigQuery.

   כניסה ל-BigQuery

2. אם רוצים להקליד את קוד PySpark ישירות, פותחים את עורך PySpark. כדי לפתוח את עורך PySpark, לוחצים על Tools > Create PySpark stored procedure.

3. כדי להגדיר אפשרויות, לוחצים על More > PySpark Options, ואז מבצעים את הפעולות הבאות:

   1. מציינים את המיקום שבו רוצים להריץ את קוד ה-PySpark.

   2. בשדה Connection, מציינים את החיבור ל-Spark.

   3. בקטע Stored procedure invocation, מציינים את מערך הנתונים שבו רוצים לאחסן את הפרוצדורות המאוחסנות הזמניות שנוצרו. אפשר להגדיר קבוצת נתונים ספציפית או לאפשר שימוש בקבוצת נתונים זמנית כדי להפעיל את קוד PySpark.

      מערך הנתונים הזמני נוצר עם המיקום שצוין בשלב הקודם. אם צוין שם של מערך נתונים, צריך לוודא שמערך הנתונים והחיבור ל-Spark נמצאים באותו מיקום.

   4. בקטע Parameters, מגדירים פרמטרים לפרוצדורה המאוחסנת. הערך של הפרמטר משמש רק במהלך הפעלות של קוד PySpark בסשן, אבל ההצהרה עצמה מאוחסנת בפרוצדורה.

   5. בקטע Advanced options, מציינים את האפשרויות של הפרוצדורה. ברשימת האפשרויות של הפרוצדורות מפורטות כל האפשרויות.

   1. בקטע Properties, מוסיפים את צמדי המפתח/ערך כדי להגדיר את המשימה. אפשר להשתמש בכל אחד מהצמדים של מפתח/ערך מתוך מאפייני Spark שנתמכים ב-Managed Service for Apache Spark.

   1. בקטע Service account settings, מגדירים את חשבון השירות בהתאמה אישית, את ה-CMEK, את מערך הנתונים של שלב הביניים ואת תיקיית Cloud Storage של שלב הביניים, שבהם ייעשה שימוש במהלך ההפעלות של קוד PySpark בסשן.

   2. לוחצים על Save.

שמירת פרוצדורה מאוחסנת ל-Spark

אחרי שיוצרים את התהליך המאוחסן באמצעות עורך PySpark, אפשר לשמור את התהליך המאוחסן. כדי לעשות את זה:

1. נכנסים לדף BigQuery במסוף Google Cloud .

   כניסה ל-BigQuery

2. בעורך השאילתות, יוצרים תהליך מאוחסן ל-Spark באמצעות Python עם עורך PySpark.

3. לוחצים על שמירה > שמירת ההליך.

4. בתיבת הדו-שיח Save stored procedure, מציינים את שם מערך הנתונים שבו רוצים לשמור את התהליך המאוחסן ואת שם התהליך המאוחסן.

5. לוחצים על Save.

   אם רוצים רק להריץ את קוד PySpark במקום לשמור אותו כפרוצדורה מאוחסנת, אפשר ללחוץ על Run במקום על Save.

שימוש בקונטיינרים בהתאמה אישית

הקונטיינר בהתאמה אישית מספק את סביבת זמן הריצה לתהליכי הדרייבר וה-executor של עומס העבודה. כדי להשתמש בקונטיינרים בהתאמה אישית, משתמשים בקוד לדוגמה הבא:

CREATE OR REPLACE PROCEDURE `PROJECT_ID`.DATASET.PROCEDURE_NAME(PROCEDURE_ARGUMENT)
  WITH CONNECTION `CONNECTION_PROJECT_ID.CONNECTION_REGION.CONNECTION_ID`
  OPTIONS (
      engine="SPARK", runtime_version="RUNTIME_VERSION",
      container_image="CONTAINER_IMAGE", main_file_uri=["MAIN_PYTHON_FILE_URI"]);
  LANGUAGE PYTHON [AS PYSPARK_CODE]

מחליפים את מה שכתוב בשדות הבאים:

• ‫PROJECT_ID: הפרויקט שבו רוצים ליצור את הפרוצדורה המאוחסנת, לדוגמה myproject.
• ‫DATASET: מערך הנתונים שבו רוצים ליצור את הפרוצדורה המאוחסנת, לדוגמה mydataset.
• ‫PROCEDURE_NAME: השם של הפרוצדורה שרוצים להריץ ב-BigQuery, למשל mysparkprocedure.
• PROCEDURE_ARGUMENT: פרמטר להזנת הארגומנטים של הקלט.

  בפרמטר הזה, מציינים את השדות הבאים:

  • ‫ARGUMENT_MODE: המצב של הארגומנט.

    הערכים התקינים כוללים את IN, OUT ו-INOUT. ערך ברירת המחדל הוא IN.

  • ‫ARGUMENT_NAME: שם הארגומנט.
  • ‫ARGUMENT_TYPE: סוג הארגומנט.
    

  לדוגמה: myproject.mydataset.mysparkproc(num INT64).

  מידע נוסף זמין בקטע העברת ערך כפרמטר IN או הפרמטרים OUT ו-INOUT במאמר הזה.

• ‫CONNECTION_PROJECT_ID: הפרויקט שמכיל את החיבור להרצת הפרוצדורה של Spark.
• ‫CONNECTION_REGION: האזור שמכיל את החיבור להרצת הפרוצדורה של Spark, לדוגמה us.
• ‫CONNECTION_ID: מזהה החיבור, לדוגמה myconnection.

  כשמציגים את פרטי החיבור במסוף Google Cloud , מזהה החיבור הוא הערך שמופיע בחלק האחרון של מזהה החיבור המלא שמוצג ב-Connection ID, לדוגמה projects/myproject/locations/connection_location/connections/myconnection.

• ‫RUNTIME_VERSION: גרסת סביבת זמן הריצה של Spark, לדוגמה 2.2.
• ‫MAIN_PYTHON_FILE_URI: הנתיב לקובץ PySpark, לדוגמה gs://mybucket/mypysparkmain.py.

  לחלופין, אם רוצים להוסיף את גוף הפרוצדורה המאוחסנת בהצהרת CREATE PROCEDURE, צריך להוסיף את PYSPARK_CODE אחרי LANGUAGE PYTHON AS, כמו שמוצג בדוגמה בקטע שימוש בקוד מוטבע במסמך הזה.

• ‫PYSPARK_CODE: ההגדרה של אפליקציית PySpark בהצהרה CREATE PROCEDURE, אם רוצים להעביר את גוף הפרוצדורה בתוך השורה.

  הערך הוא מחרוזת מילולית. אם הקוד כולל מירכאות ולוכסנים הפוכים, צריך להוסיף להם תו בריחה (escape) או לייצג אותם כמחרוזת גולמית. לדוגמה, אפשר לייצג את קוד ההחזרה "\n"; באחת מהדרכים הבאות:

  • מחרוזת במירכאות: "return \"\\n\";". גם מירכאות וגם לוכסנים הפוכים מסומנים בתווי בריחה (escape).
  • מחרוזת עם שלושה סימני מירכאות: """return "\\n";""". לוכסנים הפוכים מתווספים כתווי בריחה (escape), אבל לא מירכאות.
  • מחרוזת גולמית: r"""return "\n";""". אין צורך בתווי בריחה (escape).
  כדי ללמוד איך להוסיף קוד PySpark בתוך השורה, קראו את המאמר שימוש בקוד בתוך השורה.
• ‫CONTAINER_IMAGE: הנתיב של קובץ האימג' ב-Artifact Registry. הוא צריך להכיל רק ספריות לשימוש בפרוצדורה. אם לא מציינים, נעשה שימוש בקובץ האימג' של הקונטיינר שמשויך לגרסת סביבת זמן הריצה כברירת המחדל של המערכת.

למידע נוסף על יצירת קובץ אימג' של קונטיינר בהתאמה אישית באמצעות Spark, קראו את המאמר יצירת קובץ אימג' של קונטיינר בהתאמה אישית.

קריאה לתהליך מאוחסן ב-Spark

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

המסוף

1. נכנסים לדף BigQuery.

   כניסה ל-BigQuery

2. בחלונית הימנית, לוחצים על category (קטגוריה) Classic Explorer:

   

   אם החלונית הימנית לא מוצגת, לוחצים על הרחבת החלונית הימנית כדי לפתוח אותה.

3. בחלונית Classic Explorer, מרחיבים את הפרויקט ובוחרים את הפרוצדורה המאוחסנת ל-Spark שרוצים להריץ.

4. בחלון Stored procedure info, לוחצים על Invoke stored procedure. אפשר גם להרחיב את האפשרות View actions וללחוץ על Invoke.

5. לוחצים על Run.

6. בקטע כל התוצאות, לוחצים על הצגת התוצאות.

7. אופציונלי: בקטע Query results, פועלים לפי השלבים הבאים:

   • כדי לראות את יומני מנהלי ההתקנים של Spark, לוחצים על Execution details.

   • כדי לצפות ביומנים ב-Cloud Logging, לוחצים על Job information ואז על log בשדה Log.

   • כדי לקבל את נקודת הקצה של שרת ההיסטוריה של Spark, לוחצים על Job information ואז על Spark history server.

SQL

כדי להפעיל תהליך מאוחסן, משתמשים בהצהרה CALL PROCEDURE:

1. נכנסים לדף BigQuery במסוף Google Cloud .

   כניסה ל-BigQuery

2. בעורך השאילתות, מזינים את ההצהרה הבאה:

   CALL `PROJECT_ID`.DATASET.PROCEDURE_NAME()

3. לוחצים על הפעלה.

מידע נוסף על הרצת שאילתות זמין במאמר הרצת שאילתה אינטראקטיבית.

הערה: אפשר גם לקבל את פרטי המסנן של Cloud Logging ואת נקודת הקצה של Spark History Cluster באמצעות הפקודה bq show. כדי ללמוד איך לקבל מסנני יומנים, קראו את הקטע הצגת מסנני יומנים במסמך הזה.

שימוש בחשבון שירות בהתאמה אישית

במקום להשתמש בזהות השירות של חיבור Spark כדי לגשת לנתונים, אתם יכולים להשתמש בחשבון שירות בהתאמה אישית כדי לגשת לנתונים בקוד של Spark.

כדי להשתמש בחשבון שירות בהתאמה אישית, צריך לציין את מצב האבטחה INVOKER (באמצעות ההצהרה EXTERNAL SECURITY INVOKER) כשיוצרים פרוצדורה מאוחסנת ב-Spark, ולציין את חשבון השירות כשמפעילים את הפרוצדורה המאוחסנת.

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

כדי לגשת לקוד Spark מ-Cloud Storage ולהשתמש בו, צריך לתת את ההרשאות הנדרשות לזהות השירות של החיבור ל-Spark. אתם צריכים לתת לחשבון השירות של החיבור את הרשאת ה-IAM‏ storage.objects.get או את תפקיד ה-IAM‏ storage.objectViewer.

אם ציינתם בחיבור את Dataproc Metastore ואת Managed Service for Apache Spark Persistent History Server, אתם יכולים לתת לחשבון השירות של החיבור גישה אליהם. למידע נוסף, קראו את המאמר מתן גישה לחשבון השירות.

CREATE OR REPLACE PROCEDURE `PROJECT_ID`.DATASET.PROCEDURE_NAME(PROCEDURE_ARGUMENT)
  EXTERNAL SECURITY INVOKER
  WITH CONNECTION `CONNECTION_PROJECT_ID.CONNECTION_REGION.CONNECTION_ID`
  OPTIONS (
      engine="SPARK", runtime_version="RUNTIME_VERSION",
      main_file_uri=["MAIN_PYTHON_FILE_URI"]);
  LANGUAGE PYTHON [AS PYSPARK_CODE]

SET @@spark_proc_properties.service_account='CUSTOM_SERVICE_ACCOUNT';
CALL PROJECT_ID.DATASET_ID.PROCEDURE_NAME();

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

SET @@spark_proc_properties.staging_bucket='BUCKET_NAME';
SET @@spark_proc_properties.staging_dataset_id='DATASET';

מחליפים את מה שכתוב בשדות הבאים:

• ‫CUSTOM_SERVICE_ACCOUNT: שדה חובה. חשבון שירות בהתאמה אישית שאתם מספקים.
• ‫BUCKET_NAME: אופציונלי. הקטגוריה של Cloud Storage שמשמשת כמערכת הקבצים של אפליקציית Spark שמוגדרת כברירת מחדל. אם לא תציינו קטגוריה, תיווצר בפרויקט קטגוריית ברירת מחדל של Cloud Storage, והיא תהיה משותפת לכל המשימות שפועלות באותו פרויקט.
• ‫DATASET: אופציונלי. קבוצת הנתונים שבה יאוחסנו הנתונים הזמניים שנוצרו על ידי הפעלת הפרוצדורה. הנתונים נמחקים אחרי שהמשימה מסתיימת. אם לא מציינים קבוצת נתונים, נוצרת למשימה קבוצת נתונים זמנית כברירת מחדל.

לחשבון השירות בהתאמה אישית צריכות להיות ההרשאות הבאות:

• כדי לקרוא ולכתוב בקטגוריית ה-Staging שמשמשת כמערכת הקבצים של אפליקציית Spark שמוגדרת כברירת מחדל:

  • הרשאות storage.objects.* או תפקיד IAM מסוג roles/storage.objectAdmin בקטגוריית ה-Staging שציינתם.
  • בנוסף, ההרשאות storage.buckets.* או תפקיד ה-IAM‏ roles/storage.Admin בפרויקט, אם לא צוינה קטגוריית ה-Staging.

• (אופציונלי) כדי לקרוא ולכתוב נתונים מ-BigQuery ואליו:

  • ‫bigquery.tables.* בטבלאות ב-BigQuery.
  • ‫bigquery.readsessions.* בפרויקט.
  • התפקיד roles/bigquery.admin ב-IAM כולל את ההרשאות הקודמות.
  הערה: אם הפרוצדורה המאוחסנת כותבת נתונים לקטגוריה זמנית של Cloud Storage ואז טוענת את הנתונים של Cloud Storage ל-BigQuery, לחשבון השירות בהתאמה אישית צריכה להיות ההרשאה bigquery.jobs.create בפרויקט. במאמר תפקידים והרשאות ב-IAM תוכלו לקרוא מידע נוסף על תפקידים והרשאות ב-IAM ב-BigQuery.

• (אופציונלי) כדי לקרוא ולכתוב נתונים מ-Cloud Storage ואליו:

  • הרשאות storage.objects.* או תפקיד IAM‏ roles/storage.objectAdmin באובייקטים של Cloud Storage.

• (אופציונלי) כדי לקרוא ולכתוב למערך הנתונים של ה-staging שמשמש לפרמטרים של INOUT/OUT:

  • תפקיד IAM מסוג bigquery.tables.* או roles/bigquery.dataEditor במערך הנתונים של שלב הביניים שאתם מציינים.
  • בנוסף, ההרשאה bigquery.datasets.create או תפקיד ה-IAM‏ roles/bigquery.dataEditor בפרויקט, אם לא צוינה קבוצת נתונים להכנה.

דוגמאות לפרוצדורות מאוחסנות ב-Spark

בקטע הזה מוצגות דוגמאות ליצירת פרוצדורה מאוחסנת ל-Apache Spark.

שימוש בקובץ PySpark או JAR ב-Cloud Storage

בדוגמה הבאה מוסבר איך ליצור פרוצדורה מאוחסנת ל-Spark באמצעות החיבור my-project-id.us.my-connection וקובץ PySpark או JAR שמאוחסן בקטגוריה של Cloud Storage:

Python

CREATE PROCEDURE my_bq_project.my_dataset.spark_proc()
WITH CONNECTION `my-project-id.us.my-connection`
OPTIONS(engine="SPARK", runtime_version="2.2", main_file_uri="gs://my-bucket/my-pyspark-main.py")
LANGUAGE PYTHON

‫Java או Scala

משתמשים ב-main_file_uri כדי ליצור פרוצדורה מאוחסנת:

CREATE PROCEDURE my_bq_project.my_dataset.scala_proc_wtih_main_jar()
WITH CONNECTION `my-project-id.us.my-connection`
OPTIONS(engine="SPARK", runtime_version="2.2", main_file_uri="gs://my-bucket/my-scala-main.jar")
LANGUAGE SCALA

משתמשים ב-main_class כדי ליצור פרוצדורה מאוחסנת:

CREATE PROCEDURE my_bq_project.my_dataset.scala_proc_with_main_class()
WITH CONNECTION `my-project-id.us.my-connection`
OPTIONS(engine="SPARK", runtime_version="2.2",
main_class="com.example.wordcount", jar_uris=["gs://my-bucket/wordcount.jar"])
LANGUAGE SCALA

שימוש בקוד בתוך השורה

בדוגמה הבאה מוסבר איך ליצור פרוצדורה מאוחסנת ל-Spark באמצעות החיבור my-project-id.us.my-connection וקוד PySpark מוטבע:

CREATE OR REPLACE PROCEDURE my_bq_project.my_dataset.spark_proc()
WITH CONNECTION `my-project-id.us.my-connection`
OPTIONS(engine="SPARK", runtime_version="2.2")
LANGUAGE PYTHON AS R"""
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("spark-bigquery-demo").getOrCreate()

# Load data from BigQuery.
words = spark.read.format("bigquery") \
  .option("table", "bigquery-public-data:samples.shakespeare") \
  .load()
words.createOrReplaceTempView("words")

# Perform word count.
word_count = words.select('word', 'word_count').groupBy('word').sum('word_count').withColumnRenamed("sum(word_count)", "sum_word_count")
word_count.show()
word_count.printSchema()

# Saving the data to BigQuery
word_count.write.format("bigquery") \
  .option("writeMethod", "direct") \
  .save("wordcount_dataset.wordcount_output")
"""

העברת ערך כפרמטר קלט

בדוגמאות הבאות מוצגות שתי השיטות להעברת ערך כפרמטר קלט ב-Python:

שיטה 1: שימוש במשתני סביבה

בקוד PySpark, אפשר לקבל את הפרמטרים של הקלט של הפרוצדורה המאוחסנת ל-Spark באמצעות משתני סביבה במנהל ההתקן ובמבצעים של Spark. השם של משתנה הסביבה הוא בפורמט BIGQUERY_PROC_PARAM.PARAMETER_NAME, כאשר PARAMETER_NAME הוא השם של פרמטר הקלט. לדוגמה, אם שם הפרמטר של הקלט הוא var, שם משתנה הסביבה התואם הוא BIGQUERY_PROC_PARAM.var. הפרמטרים של הקלט מקודדים ב-JSON. בקוד PySpark, אפשר לקבל את ערך הפרמטר של הקלט במחרוזת JSON ממשתנה הסביבה, ולפענח אותו למשתנה Python.

בדוגמה הבאה אפשר לראות איך מקבלים את הערך של פרמטר קלט מסוג INT64 בקוד PySpark:

CREATE OR REPLACE PROCEDURE my_bq_project.my_dataset.spark_proc(num INT64)
WITH CONNECTION `my-project-id.us.my-connection`
OPTIONS(engine="SPARK", runtime_version="2.2")
LANGUAGE PYTHON AS R"""
from pyspark.sql import SparkSession
import os
import json

spark = SparkSession.builder.appName("spark-bigquery-demo").getOrCreate()
sc = spark.sparkContext

# Get the input parameter num in JSON string and convert to a Python variable
num = int(json.loads(os.environ["BIGQUERY_PROC_PARAM.num"]))

"""

שיטה 2: שימוש בספרייה מובנית

בקוד PySpark, אתם יכולים פשוט לייבא ספרייה מובנית ולהשתמש בה כדי לאכלס את כל סוגי הפרמטרים. כדי להעביר את הפרמטרים ל-executors, מאכלסים את הפרמטרים ב-Spark driver כמשתני Python ומעבירים את הערכים ל-executors. הספרייה המובנית תומכת ברוב סוגי הנתונים של BigQuery, חוץ מ-INTERVAL, ‏ GEOGRAPHY, ‏ NUMERIC ו-BIGNUMERIC.

סוג הנתונים ב-BigQuery	סוג הנתונים ב-Python
BOOL	bool
STRING	str
FLOAT64	float
INT64	int
BYTES	bytes
DATE	datetime.date
TIMESTAMP	datetime.datetime
TIME	datetime.time
DATETIME	datetime.datetime
Array	Array
Struct	Struct
JSON	Object
NUMERIC	לא נתמך
BIGNUMERIC	לא נתמך
INTERVAL	לא נתמך
GEOGRAPHY	לא נתמך

בדוגמה הבאה אפשר לראות איך לייבא את הספרייה המובנית ולהשתמש בה כדי לאכלס פרמטר קלט מסוג INT64 ופרמטר קלט מסוג ARRAY<STRUCT<a INT64, b STRING>> בקוד PySpark:

CREATE OR REPLACE PROCEDURE my_bq_project.my_dataset.spark_proc(num INT64, info ARRAY<STRUCT<a INT64, b STRING>>)
WITH CONNECTION `my-project-id.us.my-connection`
OPTIONS(engine="SPARK", runtime_version="2.2")
LANGUAGE PYTHON AS R"""
from pyspark.sql import SparkSession
from bigquery.spark.procedure import SparkProcParamContext

def check_in_param(x, num):
  return x['a'] + num

def main():
  spark = SparkSession.builder.appName("spark-bigquery-demo").getOrCreate()
  sc=spark.sparkContext
  spark_proc_param_context = SparkProcParamContext.getOrCreate(spark)

  # Get the input parameter num of type INT64
  num = spark_proc_param_context.num

  # Get the input parameter info of type ARRAY<STRUCT<a INT64, b STRING>>
  info = spark_proc_param_context.info

  # Pass the parameter to executors
  df = sc.parallelize(info)
  value = df.map(lambda x : check_in_param(x, num)).sum()

main()
"""

בקוד Java או Scala, אפשר לקבל את הפרמטרים של הקלט של הפרוצדורה המאוחסנת ל-Spark באמצעות משתני סביבה במנהל ההתקן ובמבצעים של Spark. הפורמט של שם משתנה הסביבה הוא BIGQUERY_PROC_PARAM.PARAMETER_NAME, כאשר PARAMETER_NAME הוא השם של פרמטר הקלט. לדוגמה, אם שם הפרמטר של הקלט הוא var, שם משתנה הסביבה התואם הוא BIGQUERY_PROC_PARAM.var. בקוד Java או Scala, אפשר לקבל את ערך הפרמטר של הקלט ממשתנה הסביבה.

בדוגמה הבאה מוסבר איך לקבל את הערך של פרמטר קלט ממשתני סביבה בקוד Scala:

val input_param = sys.env.get("BIGQUERY_PROC_PARAM.input_param").get

בדוגמה הבאה אפשר לראות איך מקבלים פרמטרים של קלט ממשתני סביבה בקוד Java:

String input_param = System.getenv("BIGQUERY_PROC_PARAM.input_param");

העברת ערכים כפרמטרים של OUT ו-INOUT

פרמטרים של פלט מחזירים את הערך מהפרוצדורה של Spark, ואילו הפרמטר INOUT מקבל ערך לפרוצדורה ומחזיר ערך מהפרוצדורה. כדי להשתמש בפרמטרים OUT ו-INOUT, צריך להוסיף את מילת המפתח OUT או INOUT לפני שם הפרמטר כשיוצרים את פרוצדורת Spark. בקוד PySpark, משתמשים בספרייה המובנית כדי להחזיר ערך כפרמטר OUT או INOUT. בדומה לפרמטרים של קלט, הספרייה המובנית תומכת ברוב סוגי הנתונים של BigQuery, מלבד INTERVAL, ‏ GEOGRAPHY,‏ NUMERIC ו-BIGNUMERIC. ערכי הסוג TIME ו-DATETIME מומרים לאזור הזמן UTC כשהם מוחזרים כפרמטרים OUT או INOUT.

CREATE OR REPLACE PROCEDURE my_bq_project.my_dataset.pyspark_proc(IN int INT64, INOUT datetime DATETIME,OUT b BOOL, OUT info ARRAY<STRUCT<a INT64, b STRING>>, OUT time TIME, OUT f FLOAT64, OUT bs BYTES, OUT date DATE, OUT ts TIMESTAMP, OUT js JSON)
WITH CONNECTION `my_bq_project.my_dataset.my_connection`
OPTIONS(engine="SPARK", runtime_version="2.2") LANGUAGE PYTHON AS
R"""
from pyspark.sql.session import SparkSession
import datetime
from bigquery.spark.procedure import SparkProcParamContext

spark = SparkSession.builder.appName("bigquery-pyspark-demo").getOrCreate()
spark_proc_param_context = SparkProcParamContext.getOrCreate(spark)

# Reading the IN and INOUT parameter values.
int = spark_proc_param_context.int
dt = spark_proc_param_context.datetime
print("IN parameter value: ", int, ", INOUT parameter value: ", dt)

# Returning the value of the OUT and INOUT parameters.
spark_proc_param_context.datetime = datetime.datetime(1970, 1, 1, 0, 20, 0, 2, tzinfo=datetime.timezone.utc)
spark_proc_param_context.b = True
spark_proc_param_context.info = [{"a":2, "b":"dd"}, {"a":2, "b":"dd"}]
spark_proc_param_context.time = datetime.time(23, 20, 50, 520000)
spark_proc_param_context.f = 20.23
spark_proc_param_context.bs = b"hello"
spark_proc_param_context.date = datetime.date(1985, 4, 12)
spark_proc_param_context.ts = datetime.datetime(1970, 1, 1, 0, 20, 0, 2, tzinfo=datetime.timezone.utc)
spark_proc_param_context.js = {"name": "Alice", "age": 30}
""";

קריאה מטבלה של Hive Metastore וכתיבת התוצאות ב-BigQuery

בדוגמה הבאה מוסבר איך להמיר טבלה של Hive Metastore ולכתוב את התוצאות ב-BigQuery:

CREATE OR REPLACE PROCEDURE my_bq_project.my_dataset.spark_proc()
WITH CONNECTION `my-project-id.us.my-connection`
OPTIONS(engine="SPARK", runtime_version="2.2")
LANGUAGE PYTHON AS R"""
from pyspark.sql import SparkSession

spark = SparkSession \
   .builder \
   .appName("Python Spark SQL Managed Service for Apache Spark Hive Metastore integration test example") \
   .enableHiveSupport() \
   .getOrCreate()

spark.sql("CREATE DATABASE IF NOT EXISTS records")

spark.sql("CREATE TABLE IF NOT EXISTS records.student (eid int, name String, score int)")

spark.sql("INSERT INTO records.student VALUES (1000000, 'AlicesChen', 10000)")

df = spark.sql("SELECT * FROM records.student")

df.write.format("bigquery") \
  .option("writeMethod", "direct") \
  .save("records_dataset.student")
"""

הצגת מסנני היומן

אחרי שקוראים לפרוצדורה מאוחסנת ל-Spark, אפשר לראות את פרטי היומן. כדי לקבל את פרטי המסנן של Cloud Logging ואת נקודת הקצה של Spark History Cluster, משתמשים בפקודה bq show. פרטי המסנן זמינים בשדה SparkStatistics של המשימה הצאצא. כדי לקבל מסנני יומנים, פועלים לפי השלבים הבאים:

1. נכנסים לדף BigQuery.

   כניסה ל-BigQuery

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

   bq ls -j --parent_job_id=$parent_job_id

   כדי ללמוד איך לקבל את מזהה המשימה, קראו את המאמר הצגת פרטי המשימה.

   הפלט אמור להיראות כך:

                   jobId                         Job Type     State       Start Time         Duration
   ---------------------------------------------- ---------   ---------  ---------------  ----------------
   script_job_90fb26c32329679c139befcc638a7e71_0   query      SUCCESS   07 Sep 18:00:27   0:05:15.052000

3. מזהים את ה-jobId של הפרוצדורה המאוחסנת ומשתמשים בפקודה bq show כדי לראות את פרטי המשימה:

   bq show --format=prettyjson --job $child_job_id

   מעתיקים את השדה sparkStatistics כי תצטרכו אותו בשלב אחר.

   הפלט אמור להיראות כך:

   {
   "configuration": {...}
   …
   "statistics": {
    …
     "query": {
       "sparkStatistics": {
         "loggingInfo": {
           "projectId": "myproject",
           "resourceType": "myresource"
         },
         "sparkJobId": "script-job-90f0",
         "sparkJobLocation": "us-central1"
       },
       …
     }
   }
   }

4. ב-Logging, יוצרים מסנני יומנים עם השדות של SparkStatistics:

   resource.type = sparkStatistics.loggingInfo.resourceType
   resource.labels.resource_container=sparkStatistics.loggingInfo.projectId
   resource.labels.spark_job_id=sparkStatistics.sparkJobId
   resource.labels.location=sparkStatistics.sparkJobLocation

   היומנים נכתבים במשאב במעקב bigquery.googleapis.com/SparkJob. היומנים מתויגים לפי הרכיבים INFO,‏ DRIVER ו-EXECUTOR. כדי לסנן יומנים ממנהל ההתקן של Spark, מוסיפים את הרכיב labels.component = "DRIVER" למסנני היומנים. כדי לסנן יומנים מה-Spark executor, מוסיפים את הרכיב labels.component = "EXECUTOR" למסנני היומנים.

שימוש במפתח הצפנה בניהול הלקוח

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

אם הפרוצדורה המאוחסנת משתמשת במצב האבטחה INVOKER, צריך לציין את ה-CMEK באמצעות משתנה המערכת של SQL כשקוראים לפרוצדורה. אחרת, אפשר לציין את ה-CMEK דרך החיבור שמשויך לפרוצדורה המאוחסנת.

כדי לציין את ה-CMEK דרך החיבור כשיוצרים פרוצדורה מאוחסנת ב-Spark, משתמשים בקוד לדוגמה הבא:

bq mk --connection --connection_type='SPARK' \
 --properties='{"kms_key_name"="projects/PROJECT_ID/locations/LOCATION/keyRings/KEY_RING_NAME/cryptoKeys/KMS_KEY_NAME"}' \
 --project_id=PROJECT_ID \
 --location=LOCATION \
 CONNECTION_NAME

כדי לציין CMEK באמצעות משתנה המערכת של SQL כשקוראים לפרוצדורה, משתמשים בקוד הבא לדוגמה:

SET @@spark_proc_properties.service_account='CUSTOM_SERVICE_ACCOUNT';
SET @@spark_proc_properties.kms_key_name='projects/PROJECT_ID/locations/LOCATION/keyRings/KEY_RING_NAME/cryptoKeys/KMS_KEY_NAME;
CALL PROJECT_ID.DATASET_ID.PROCEDURE_NAME();

שימוש ב-VPC Service Controls

בעזרת VPC Service Controls אתם יכולים להגדיר מתחם אבטחה היקפית כדי להגן מפני זליגת נתונים. כדי להשתמש ב-VPC Service Controls עם פרוצדורת Spark לאבטחה נוספת, קודם צריך ליצור גבולות גזרה לשירות.

כדי להגן באופן מלא על משימות הפרוצדורה של Spark, צריך להוסיף את ממשקי ה-API הבאים לגבולות גזרה לשירות:

• ‫BigQuery API‏ (bigquery.googleapis.com)
• ‫Cloud Logging API‏ (logging.googleapis.com)
• ‫Cloud Storage API‏ (storage.googleapis.com), אם משתמשים ב-Cloud Storage
• ‫Artifact Registry API‏ (artifactregistry.googleapis.com) או Container Registry API‏ (containerregistry.googleapis.com), אם משתמשים בקונטיינר בהתאמה אישית
• ‫Dataproc Metastore API (metastore.googleapis.com) ו-Cloud Run Admin API (run.googleapis.com), אם משתמשים ב-Dataproc Metastore

מוסיפים להיקף את פרויקט השאילתה של פרוצדורת Spark. להוסיף להיקף פרויקטים אחרים שמארחים את הקוד או הנתונים של Spark.

שיטות מומלצות

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

• כשיוצרים פרוצדורת Spark לשימוש בסביבת ייצור, Google ממליצה לציין גרסת זמן ריצה. כאן תוכלו למצוא רשימה של גרסאות זמן הריצה שנתמכות ב-Managed Service for Apache Spark. אנחנו ממליצים להשתמש בגרסת Long-Term Support ‏ (LTS).

• כשמציינים קונטיינר בהתאמה אישית בפרוצדורת Spark, מומלץ להשתמש ב-Artifact Registry ובסטרימינג של קובצי אימג'.

• כדי לשפר את הביצועים, אפשר לציין מאפיינים של הקצאת משאבים בפרוצדורה של Spark. הפרוצדורות המאוחסנות של Spark תומכות ברשימה של מאפייני הקצאת משאבים, כמו Managed Service for Apache Spark.

מגבלות

• כדי להתחבר ל-Dataproc Metastore, אפשר להשתמש רק בפרוטוקול נקודת קצה (endpoint) של gRPC. אין תמיכה בסוגים אחרים של Hive Metastore.
• מפתחות הצפנה בניהול הלקוח (CMEK) זמינים רק כשלקוחות יוצרים פרוצדורות Spark באזור יחיד. אין תמיכה במפתחות CMEK באזור גלובלי ובמפתחות CMEK במספר אזורים, לדוגמה EU או US.
• אפשר להעביר פרמטרים של פלט רק ב-PySpark.
• אם קבוצת הנתונים שמשויכת לפרוצדורה המאוחסנת של Spark משוכפלת לאזור יעד באמצעות שכפול של קבוצת נתונים בין אזורים, אפשר לשלוח שאילתות לפרוצדורה המאוחסנת רק באזור שבו היא נוצרה.
• ב-Spark אין תמיכה בגישה לנקודות קצה (endpoints) של HTTP ברשת הפרטית של VPC Service Controls.

מכסות ומגבלות

למידע על מכסות ומגבלות, ראו מכסות ומגבלות של פרוצדורות מאוחסנות ל-Spark.

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

• איך מציגים פרוצדורה מאוחסנת
• איך מוחקים פרוצדורה מאוחסנת
• איך עובדים עם פרוצדורה מאוחסנת ב-SQL