הרצת קוד PySpark ב-notebooks של BigQuery Studio

אתם יכולים ליצור סשן Spark ב-notebook של BigQuery Studio כדי להשתמש בקוד PySpark לניתוח ולטרנספורמציה של הנתונים בקנה מידה גדול יותר. השילוב המנוהל הזה מאפשר להוסיף חבילות Python באופן דינמי ולעבד מערכי נתונים גדולים בלי לצאת מסביבת ה-Notebook.

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

אם עדיין לא עשיתם את זה, יוצרים פרויקט ב- Google Cloud וקטגוריה ב-Cloud Storage.

  1. הגדרת הפרויקט

    1. נכנסים לחשבון Google Cloud . אנחנו ממליצים למשתמשים חדשים ב- Google Cloud ליצור חשבון כדי שיוכלו להעריך את הביצועים של המוצרים שלנו בתרחישים מהעולם האמיתי. לקוחות חדשים מקבלים בחינם גם קרדיט בשווי 300$ להרצה, לבדיקה ולפריסה של עומסי העבודה.

    2. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

      Roles required to select or create a project

      • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
      • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

      Go to project selector

    3. Enable the Managed Service for Apache Spark, BigQuery, and Cloud Storage APIs, if any are not already enabled.

      Roles required to enable APIs

      To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

      Enable the APIs

    4. In the Google Cloud console, on the project selector page, select or create a Google Cloud project.

      Roles required to select or create a project

      • Select a project: Selecting a project doesn't require a specific IAM role—you can select any project that you've been granted a role on.
      • Create a project: To create a project, you need the Project Creator role (roles/resourcemanager.projectCreator), which contains the resourcemanager.projects.create permission. Learn how to grant roles.

      Go to project selector

    5. Enable the Managed Service for Apache Spark, BigQuery, and Cloud Storage APIs, if any are not already enabled.

      Roles required to enable APIs

      To enable APIs, you need the serviceusage.services.enable permission. If you created the project, then you likely already have this permission through the Owner role (roles/owner). Otherwise, you can get this permission through the Service Usage Admin role (roles/serviceusage.serviceUsageAdmin). Learn how to grant roles.

      Enable the APIs

      6.

  2. יוצרים קטגוריה של Cloud Storage בפרויקט, אם אין לכם קטגוריה שאתם יכולים להשתמש בה.

  3. הגדרת ה-Notebook

תמחור

למידע על התמחור של סביבות זמן ריצה של Notebook ב-BigQuery

פתיחת Notebook של Python ב-BigQuery Studio

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

    כניסה ל-BigQuery

  2. בסרגל הכרטיסיות של חלונית הפרטים, לוחצים על החץ ליד הסימן + ואז על Notebook.

יצירת סשן Spark ב-notebook של BigQuery Studio

אתם יכולים להשתמש ב-notebook של Python ב-BigQuery Studio כדי ליצור סשן אינטראקטיבי של Spark Connect. לכל Notebook ב-BigQuery Studio יכול להיות רק סשן Spark פעיל אחד שמשויך אליו.

אפשר ליצור סשן Spark ב-Notebook של Python ב-BigQuery Studio בדרכים הבאות:

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

סשן אחד

כדי ליצור סשן Spark ב-notebook חדש:

  1. בסרגל הכרטיסיות של חלונית העריכה, לוחצים על החץ לתפריט הנפתח לצד הסימן + ואז לוחצים על Notebook.

    צילום מסך של ממשק BigQuery עם הכפתור "+" ליצירת Notebook חדש.

  2. מעתיקים ומריצים את הקוד הבא בתא של ה-notebook כדי להגדיר וליצור סשן בסיסי של Spark.

from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session

import pyspark.sql.functions as f

session = Session()

# Create the Spark session.
spark = (
   DataprocSparkSession.builder
     .appName("APP_NAME")
     .dataprocSessionConfig(session)
     .getOrCreate()
)

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

  • ‫APP_NAME: שם אופציונלי לסשן.
  • הגדרות אופציונליות לסשן: אתם יכולים להוסיף הגדרות של Managed Service for Apache Spark API Session כדי להתאים אישית את הסשן. הנה כמה דוגמאות:
    • RuntimeConfig:
      עזרה בקוד שמוצגות בה אפשרויות של session.runtime.config.
      • session.runtime_config.properties={spark.property.key1:VALUE_1,...,spark.property.keyN:VALUE_N}
      • session.runtime_config.container_image = path/to/container/image
    • EnvironmentConfig:
      עזרה לגבי קוד שמוצגות בה אפשרויות של session-environment-config-execution-config.
      • session.environment_config.execution_config.subnetwork_uri = "SUBNET_NAME"
      • session.environment_config.execution_config.ttl = {"seconds": VALUE}
      • session.environment_config.execution_config.service_account = SERVICE_ACCOUNT

סשן Spark עם תבנית

אתם יכולים להזין ולהריץ את הקוד בתא של Notebook כדי ליצור סשן Spark על סמך תבנית סשן קיימת. כל הגדרות אישיות של session שתספקו בקוד של ה-notebook יחליפו את ההגדרות הזהות שנקבעו בתבנית הסשן.

כדי להתחיל במהירות, אפשר להשתמש בתבנית Query using Spark כדי לאכלס מראש את ה-notebook בקוד של תבנית סשן Spark:

  1. בסרגל הכרטיסיות של חלונית העריכה, לוחצים על החץ לתפריט הנפתח לצד הסימן + ואז לוחצים על Notebook.
    צילום מסך של ממשק BigQuery עם הכפתור "+" ליצירת Notebook חדש.
  2. בקטע Start with a template, לוחצים על Query using Spark ואז על Use template כדי להוסיף את הקוד ל-notebook.
    אפשרויות בממשק המשתמש של BigQuery כדי להתחיל עם תבנית
  3. מציינים את המשתנים כמו שמוסבר בהערות.
  4. אתם יכולים למחוק את כל התאים הנוספים של הקוד לדוגמה שהוספתם ל-Notebook.
from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session
session = Session()
project_id = "PROJECT_ID"
location = "LOCATION"
# Configure the session with an existing session template.
session_template = "SESSION_TEMPLATE"
session.session_template = f"projects/{project_id}/locations/{location}/sessionTemplates/{session_template}"
# Create the Spark session.
spark = (
   DataprocSparkSession.builder
     .appName("APP_NAME")
     .dataprocSessionConfig(session)
     .getOrCreate()
)

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

  • ‫PROJECT_ID: מזהה הפרויקט, שמופיע בקטע Project info במרכז הבקרה של מסוףGoogle Cloud .
  • ‫LOCATION: האזור ב-Compute Engine שבו יפעל הסשן של ה-Notebook. אם לא תציינו אזור, המערכת תשתמש באזור של ה-VM שיוצר את ה-Notebook.

  • ‫SESSION_TEMPLATE: השם של תבנית הסשן האינטראקטיבי הקיימת. הגדרות הסשן נלקחות מהתבנית. בתבנית צריך לציין גם את ההגדרות הבאות:

    • גרסת זמן ריצה 2.3+

    • סוג ה-Notebook: ‏Spark Connect

      דוגמה:

      צילום מסך שבו מוצגות ההגדרות הנדרשות של Spark Connect.

  • ‫APP_NAME: שם אופציונלי לסשן.

כתיבה והרצה של קוד PySpark ב-notebook ב-BigQuery Studio

אחרי שיוצרים סשן Spark ב-notebook, משתמשים בסשן כדי להריץ קוד Spark notebook ב-notebook.

תמיכה ב-Spark Connect PySpark API: הסשן של Spark Connect ב-Notebook תומך ברוב ממשקי ה-API של PySpark, כולל DataFrame,‏ Functions ו-Column, אבל לא תומך ב-SparkContext, ב-RDD ובממשקי API אחרים של PySpark. מידע נוסף זמין במאמר מה נתמך ב-Spark 3.5.

כתיבה ישירה ב-Notebook של Spark Connect: סשנים של Spark ב-Notebook של BigQuery Studio מגדירים מראש את מחבר Spark BigQuery כדי לבצע כתיבה ישירה של נתונים. בשיטת הכתיבה DIRECT נעשה שימוש ב-BigQuery Storage Write API, שכותב נתונים ישירות ל-BigQuery. בשיטת הכתיבה INDIRECT, שהיא ברירת המחדל של אצוות Managed Service for Apache Spark, הנתונים נכתבים לקטגוריה זמנית של Cloud Storage ואז ל-BigQuery (מידע נוסף על כתיבה בשיטת INDIRECT זמין במאמר קריאה וכתיבה של נתונים מ-BigQuery ואליו).

ממשקי API ספציפיים ל-Managed Service for Apache Spark: השירות המנוהל ל-Apache Spark מפשט את הוספת החבילות של PyPI באופן דינמי לסשן Spark, על ידי הרחבת ה-method‏ addArtifacts. אפשר לציין את הרשימה בפורמט version-scheme (בדומה ל-pip install). הפעולה הזו מורה לשרת Spark Connect להתקין חבילות ואת יחסי התלות שלהן בכל הצמתים באשכול, כדי שהן יהיו זמינות ל-workers בשביל ה-UDF שלכם.

דוגמה להתקנה של גרסת textdistance שצוינה ושל ספריות random2 תואמות עדכניות באשכול, כדי לאפשר לפונקציות UDF שמשתמשות ב-textdistance וב-random2 לפעול בצומתי עובדים.

spark.addArtifacts("textdistance==4.6.1", "random2", pypi=True)

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

בדוגמה הבאה, אם מזינים DataprocSparkSession ומחזיקים את הסמן מעל שם המחלקה, מוצגת עזרה להשלמת הקוד ולתיעוד.

דוגמאות לטיפים בנושא תיעוד קוד והשלמת קוד.

דוגמאות ל-PySpark ב-notebook של BigQuery Studio

בקטע הזה מופיעות דוגמאות ל-notebooks של Python ב-BigQuery Studio עם קוד PySpark, כדי לבצע את המשימות הבאות:

ספירת מילים

בדוגמה הבאה ב-PySpark נוצר סשן Spark, ואז נספרות המופעים של מילים במערך נתונים ציבורי של bigquery-public-data.samples.shakespeare.

# Basic wordcount example
from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session
import pyspark.sql.functions as f
session = Session()

# Create the Spark session.
spark = (
   DataprocSparkSession.builder
     .appName("APP_NAME")
     .dataprocSessionConfig(session)
     .getOrCreate()
)
# Run a wordcount on the public Shakespeare dataset.
df = spark.read.format("bigquery").option("table", "bigquery-public-data.samples.shakespeare").load()
words_df = df.select(f.explode(f.split(f.col("word"), " ")).alias("word"))
word_counts_df = words_df.filter(f.col("word") != "").groupBy("word").agg(f.count("*").alias("count")).orderBy("word")
word_counts_df.show()

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

  • ‫APP_NAME: שם אופציונלי לסשן.

פלט:

בפלט של התא מופיעה דוגמה לפלט של ספירת המילים. כדי לראות את פרטי הסשן האינטראקטיבי במסוף Google Cloud , לוחצים על הקישור Interactive Session Detail View. נפתחת חלונית פרטים של Spark עם הכרטיסיות Logs,‏ Executors,‏ SQL ו-Jobs (כולל ציר זמן של אירועים). כדי לעקוב אחרי הסשן ב-Spark, לוחצים על View Spark UI בדף פרטי הסשן.

כפתור לצפייה בממשק המשתמש של Spark בדף פרטי הסשן במסוף 
Interactive Session Detail View: LINK
+------------+-----+
|        word|count|
+------------+-----+
|           '|   42|
|       ''All|    1|
|     ''Among|    1|
|       ''And|    1|
|       ''But|    1|
|    ''Gamut'|    1|
|       ''How|    1|
|        ''Lo|    1|
|      ''Look|    1|
|        ''My|    1|
|       ''Now|    1|
|         ''O|    1|
|      ''Od's|    1|
|       ''The|    1|
|       ''Tis|    4|
|      ''When|    1|
|       ''tis|    1|
|      ''twas|    1|
|          'A|   10|
|'ARTEMIDORUS|    1|
+------------+-----+
only showing top 20 rows

טבלת Iceberg

מריצים קוד PySpark כדי ליצור טבלת Iceberg עם מטא-נתונים של קטלוג ייעודי לזמן ריצה של Lakehouse

בדוגמת הקוד הבאה נוצר sample_iceberg_table עם מטא-נתונים של טבלה שמאוחסנים בקטלוג ייעודי לזמן ריצה של Lakehouse, ואז מתבצעת שאילתה על הטבלה.

from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session
# Create the Dataproc Serverless session.
session = Session()
# Set the session configuration for BigLake Metastore with the Iceberg environment.
project_id = "PROJECT_ID"
region = "REGION"
subnet_name = "SUBNET_NAME"
location = "LOCATION"
session.environment_config.execution_config.subnetwork_uri = f"{subnet_name}"
warehouse_dir = "gs://BUCKET/WAREHOUSE_DIRECTORY"
catalog = "CATALOG"
namespace = "NAMESPACE"
session.runtime_config.properties[f"spark.sql.catalog.{catalog}"] = "org.apache.iceberg.spark.SparkCatalog"
session.runtime_config.properties[f"spark.sql.catalog.{catalog}.catalog-impl"] = "org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog"
session.runtime_config.properties[f"spark.sql.catalog.{catalog}.gcp_project"] = f"{project_id}"
session.runtime_config.properties[f"spark.sql.catalog.{catalog}.gcp_location"] = f"{location}"
session.runtime_config.properties[f"spark.sql.catalog.{catalog}.warehouse"] = f"{warehouse_dir}"
# Create the Spark Connect session.
spark = (
   DataprocSparkSession.builder
     .appName("APP_NAME")
     .dataprocSessionConfig(session)
     .getOrCreate()
)
# Create the namespace in BigQuery.
spark.sql(f"USE `{catalog}`;")
spark.sql(f"CREATE NAMESPACE IF NOT EXISTS `{namespace}`;")
spark.sql(f"USE `{namespace}`;")
# Create the Iceberg table.
spark.sql("DROP TABLE IF EXISTS `sample_iceberg_table`");
spark.sql("CREATE TABLE sample_iceberg_table (id int, data string) USING ICEBERG;")
spark.sql("DESCRIBE sample_iceberg_table;")
# Insert table data and query the table.
spark.sql("INSERT INTO sample_iceberg_table VALUES (1, \"first row\");")
# Alter table, then query and display table data and schema.
spark.sql("ALTER TABLE sample_iceberg_table ADD COLUMNS (newDoubleCol double);")
spark.sql("DESCRIBE sample_iceberg_table;")
df = spark.sql("SELECT * FROM sample_iceberg_table")
df.show()
df.printSchema()

הערות:

  • ‫PROJECT_ID: מזהה הפרויקט, שמופיע בקטע Project info במרכז הבקרה של מסוףGoogle Cloud .
  • ‫REGION ו-SUBNET_NAME: מציינים את האזור של Compute Engine ואת השם של רשת המשנה באזור של הסשן. ‫Managed Service for Apache Spark מפעיל את גישה פרטית ל-Google‏ (PGA) ברשת המשנה שצוינה.
  • ‫LOCATION: ברירת המחדל של BigQuery_metastore_config.location ושל spark.sql.catalog.{catalog}.gcp_location היא US, אבל אפשר לבחור כל מיקום נתמך ב-BigQuery.
  • ‫BUCKET ו-WAREHOUSE_DIRECTORY: הקטגוריה והתיקייה של Cloud Storage שמשמשות לספריית המחסן של Iceberg.
  • ‫CATALOG ו-NAMESPACE: השילוב של שם הקטלוג ומרחב השמות של Iceberg מזהה את הטבלה של Iceberg‏ (catalog.namespace.table_name).
  • ‫APP_NAME: שם אופציונלי לסשן.

בפלט של התא מופיעה רשימה של sample_iceberg_table עם העמודה שנוספה, וקישור לדף פרטי סשן אינטראקטיבי במסוף Google Cloud . נפתחת חלונית פרטים של Spark עם הכרטיסיות Logs,‏ Executors,‏ SQL ו-Jobs (כולל ציר זמן של אירועים). כדי לעקוב אחרי הסשן ב-Spark, אתם יכולים ללחוץ על View Spark UI בדף הפרטים של הסשן.

Interactive Session Detail View: LINK
+---+---------+------------+
| id|     data|newDoubleCol|
+---+---------+------------+
|  1|first row|        NULL|
+---+---------+------------+

root
 |-- id: integer (nullable = true)
 |-- data: string (nullable = true)
 |-- newDoubleCol: double (nullable = true)

הצגת פרטי הטבלה ב-BigQuery

כדי לבדוק את פרטי הטבלה של Iceberg ב-BigQuery:

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

    כניסה ל-BigQuery

  2. בחלונית המשאבים של הפרויקט, לוחצים על הפרויקט ואז על מרחב השמות כדי להציג את הטבלה sample_iceberg_table. לוחצים על הטבלה Details כדי לראות את המידע של Open Catalog Table Configuration.

    פורמטי הקלט והפלט הם הפורמטים הרגילים של Hadoop InputFormat ו-OutputFormat שבהם משתמשים ב-Iceberg.

    מטא-נתונים של טבלת Iceberg שמופיעים בממשק המשתמש של BigQuery

דוגמאות נוספות

יוצרים Spark DataFrame (sdf) מ-Pandas DataFrame (df).

sdf = spark.createDataFrame(df)
sdf.show()

הרצת צבירות ב-Spark DataFrames.

from pyspark.sql import functions as f

sdf.groupby("segment").agg(
   f.mean("total_spend_per_user").alias("avg_order_value"),
   f.approx_count_distinct("user_id").alias("unique_customers")
).show()

קוראים מ-BigQuery באמצעות המחבר Spark-BigQuery.

spark.conf.set("viewsEnabled","true")
spark.conf.set("materializationDataset","my-bigquery-dataset")

sdf = spark.read.format('bigquery') \
 .load(query)

כתיבת קוד Spark באמצעות Gemini Code Assist

אתם יכולים לבקש מ-Gemini Code Assist ליצור קוד PySpark ב-notebook. ‫Gemini Code Assist מאחזר ומשתמש בטבלאות רלוונטיות של BigQuery ו-Dataproc Metastore ובסכימות שלהן כדי ליצור תשובה בקוד.

כדי ליצור קוד באמצעות Gemini Code Assist ב-notebook:

  1. כדי להוסיף תא קוד חדש, לוחצים על ‎+ Code בסרגל הכלים. תא הקוד החדש מציג את Start coding or generate with AI. לוחצים על יצירה.

  2. בעורך, כותבים הנחיה בשפה טבעית ולוחצים על enter. חשוב לכלול בהנחיה את מילת המפתח spark או pyspark.

    פרומפט לדוגמה: 

    create a spark dataframe from order_items and filter to orders created in 2024

    פלט לדוגמה:

    spark.read.format("bigquery").option("table", "sqlgen-testing.pysparkeval_ecommerce.order_items").load().filter("year(created_at) = 2024").createOrReplaceTempView("order_items")
df = spark.sql("SELECT * FROM order_items")

טיפים ליצירת קוד באמצעות Gemini Code Assist

  • כדי לאפשר ל-Gemini Code Assist לאחזר טבלאות וסכימות רלוונטיות, צריך להפעיל את סנכרון Data Catalog במכונות של Dataproc Metastore.

  • ודאו שלחשבון המשתמש שלכם יש גישה לטבלאות השאילתות ב-Data Catalog. כדי לעשות את זה, מקצים את התפקיד DataCatalog.Viewer.

סיום הסשן ב-Spark

כדי להפסיק את הסשן של Spark Connect ב-notebook ב-BigQuery Studio, אתם יכולים לבצע כל אחת מהפעולות הבאות:

  • מריצים את spark.stop() בתא של Notebook.
  • סיום זמן הריצה ב-Notebook:
    1. לוחצים על בורר סביבת זמן הריצה ואז על Manage sessions.
      ניהול הסשנים שנבחרו
    2. בתיבת הדו-שיח Active sessions, לוחצים על סמל הסיום ואז על Terminate.
      סיום הסשן שנבחר בתיבת הדו-שיח "סשנים פעילים"

תזמור קוד של Notebook ב-BigQuery Studio

אתם יכולים לתזמר את הקוד של ה-Notebook ב-BigQuery Studio בדרכים הבאות:

תזמון קוד של Notebook ממסוף Google Cloud

אפשר לתזמן קוד של Notebook בדרכים הבאות:

הרצת קוד של Notebook כעומס עבודה באצווה

כדי להריץ קוד של BigQuery Studio Notebook כעומס עבודה באצווה:

  1. מורידים את קוד ה-Notebook לקובץ בטרמינל מקומי או ב-Cloud Shell.

    1. במסוף Google Cloud , פותחים את ה-notebook בחלונית Explorer בדף BigQuery Studio.

    2. כדי להרחיב את סרגל התפריטים, לוחצים על keyboard_arrow_down הצגה או הסתרה של הכותרת.

    3. לוחצים על קובץ > הורדה, ואז על Download.py.

      התפריט "קובץ" > "הורדה" בדף Explorer.

  2. יוצרים requirements.txt.

    1. מתקינים את pipreqs בספרייה שבה שמרתם את הקובץ .py. 
      pip install pipreqs

    2. מריצים את pipreqs כדי ליצור את requirements.txt. 

      pipreqs filename.py

    3. משתמשים ב-Google Cloud CLI כדי להעתיק את הקובץ המקומי requirements.txt לקטגוריה ב-Cloud Storage.

      gcloud storage cp requirements.txt gs://BUCKET/

  3. מעדכנים את קוד הסשן של Spark על ידי עריכת קובץ ה-.py שהורדתם.

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

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

      דוגמה:

      • מסירים מהקוד את שורת ההגדרה הבאה של רשת המשנה של הסשן: 

        session.environment_config.execution_config.subnetwork_uri = "{subnet_name}"

      • כשמריצים את עומס העבודה באצווה, משתמשים בדגל --subnet כדי לציין את רשת המשנה.

        gcloud dataproc batches submit pyspark \
--subnet=SUBNET_NAME

    3. משתמשים בקטע קוד פשוט ליצירת סשן.

      • קוד לדוגמה של notebook שהורד לפני פישוט.

        from google.cloud.dataproc_spark_connect import DataprocSparkSession
from google.cloud.dataproc_v1 import Session

        
session = Session()
spark = DataprocSparkSession \
    .builder \
    .appName("CustomSparkSession")
    .dataprocSessionConfig(session) \
    .getOrCreate()

      • קוד עומס העבודה של Batch אחרי פישוט.

        from pyspark.sql import SparkSession

        
spark = SparkSession \
.builder \
.getOrCreate()

  4. מריצים את עומס העבודה באצווה.

    1. הוראות מפורטות זמינות במאמר שליחת עומס העבודה באצווה של Spark.

      • חשוב לכלול את הדגל ‎--deps-bucket כדי להפנות לקטגוריה של Cloud Storage שמכילה את הקובץ requirements.txt.

        דוגמה:

      gcloud dataproc batches submit pyspark FILENAME.py \
    --region=REGION \
    --deps-bucket=BUCKET \
    --version=2.3

      הערות:

      • ‫FILENAME: השם של קובץ הקוד של ה-notebook שהורדתם וערכתם.
      • ‫REGION: האזור ב-Compute Engine שבו נמצא האשכול.
      • ‫BUCKET השם של הקטגוריה ב-Cloud Storage שמכילה את קובץ ה-requirements.txt.
      • ‫--version: נבחרה גרסת זמן הריצה של Spark‏ 2.3 כדי להריץ את עומס העבודה באצווה.

  5. מבצעים commit לקוד.

    1. אחרי שבודקים את הקוד של עומס העבודה באצווה, אפשר להעביר את הקובץ .ipynb או .py למאגר באמצעות לקוח git, כמו GitHub,‏ GitLab או Bitbucket, כחלק מצינור עיבוד הנתונים של CI/CD.

  6. תזמון עומס העבודה באצווה באמצעות Managed Service for Apache Airflow.

    1. הוראות מפורטות זמינות במאמר הפעלת עומסי עבודה של Managed Service for Apache Spark באמצעות Managed Airflow.

פתרון בעיות שקשורות לשגיאות ב-Notebook

אם מתרחשת שגיאה בתא שמכיל קוד Spark, אפשר לפתור אותה על ידי לחיצה על הקישור Interactive Session Detail View בפלט של התא (ראו דוגמאות לטבלאות Wordcount ו-Iceberg).

בעיות מוכרות ופתרונות

שגיאה: זמן ריצה של Notebook שנוצר באמצעות גרסה 3.10 של Python עלול לגרום לשגיאה PYTHON_VERSION_MISMATCH כשמנסים להתחבר לסשן של Spark.

פתרון: יוצרים מחדש את סביבת זמן הריצה עם גרסה 3.11 של Python.

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