PySpark-Code in BigQuery Studio-Notebooks ausführen

Sie können eine Spark-Sitzung in einem BigQuery Studio-Notebook erstellen, um PySpark-Code zu verwenden, um Ihre Daten in größerem Umfang zu analysieren und zu transformieren. Mit dieser verwalteten Integration können Sie Python-Pakete dynamisch hinzufügen und große Datasets verarbeiten, ohne die Notebook-Umgebung zu verlassen.

Hinweis

Erstellen Sie ein Google Cloud -Projekt und einen Cloud Storage-Bucket, falls noch nicht geschehen.

1. Projekt einrichten

   Melden Sie sich in Ihrem Google Cloud -Konto an. Wenn Sie mit Google Cloudnoch nicht vertraut sind, erstellen Sie einfach ein Konto, um die Leistungsfähigkeit unserer Produkte in der Praxis sehen und bewerten zu können. Neukunden erhalten außerdem ein Guthaben von 300 $, um Arbeitslasten auszuführen, zu testen und bereitzustellen.

   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.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

   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

   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.

   Note: If you don't plan to keep the resources that you create in this procedure, create a project instead of selecting an existing project. After you finish these steps, you can delete the project, removing all resources associated with the project.

   Go to project selector

   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

2. Erstellen Sie einen Cloud Storage-Bucket in Ihrem Projekt, wenn Sie keinen Bucket haben, den Sie verwenden können.

3. Notebook einrichten

   • Notebook-Anmeldedaten: Standardmäßig werden für Ihre Notebook-Sitzung Ihre Nutzeranmeldedaten verwendet. Alternativ können die Anmeldedaten des Sitzungsdienstkontos verwendet werden.
     • Nutzeranmeldedaten: Ihr Nutzerkonto muss die folgenden IAM-Rollen (Identity and Access Management) haben:
       • Managed Service for Apache Spark Editor (Rolle roles/dataproc.editor)
       • BigQuery Studio User (Rolle roles/bigquery.studioUser)
       • Rolle „Dienstkontonutzer“ (roles/iam.serviceAccountUser) für das Sitzungsdienstkonto. Diese Rolle enthält die erforderliche Berechtigung iam.serviceAccounts.actAs, um die Identität des Dienstkontos zu übernehmen.
     • Dienstkonto-Anmeldedaten: Wenn Sie für Ihre Notebook-Sitzung Dienstkonto-Anmeldedaten anstelle von Nutzeranmeldedaten angeben möchten, muss das Sitzungsdienstkonto die folgende Rolle haben:
       • Dataproc-Worker (Rolle roles/dataproc.worker)
   • Notebook-Laufzeit: Ihr Notebook verwendet eine Standardlaufzeit der Gemini Enterprise Agent Platform, sofern Sie keine andere Laufzeit auswählen. Wenn Sie eine eigene Laufzeit definieren möchten, erstellen Sie die Laufzeit in der Google Cloud Console auf der Seite Laufzeiten. Hinweis: Wenn Sie die NumPy-Bibliothek verwenden, müssen Sie in der Notebook-Laufzeit NumPy-Version 1.26 verwenden, die von Spark 3.5 unterstützt wird.

Preise

Preisinformationen finden Sie unter BigQuery-Preise für Notebook-Laufzeiten.

Python-Notebook in BigQuery Studio öffnen

1. Rufen Sie in der Google Cloud Console die Seite BigQuery auf.

   BigQuery aufrufen

2. Klicken Sie in der Tableiste des Detailbereichs auf den -Pfeil neben dem +-Zeichen und dann auf Notebook.

Spark-Sitzung in einem BigQuery Studio-Notebook erstellen

Sie können ein BigQuery Studio-Python-Notebook verwenden, um eine interaktive Spark Connect-Sitzung zu erstellen. Jedem BigQuery Studio-Notebook kann nur eine aktive Spark-Sitzung zugeordnet sein.

Sie haben folgende Möglichkeiten, eine Spark-Sitzung in einem BigQuery Studio-Python-Notebook zu erstellen:

• Konfigurieren und erstellen Sie eine einzelne Sitzung im Notebook.
• Konfigurieren Sie eine Spark-Sitzung in einer interaktiven Sitzungsvorlage und verwenden Sie die Vorlage dann, um eine Sitzung im Notebook zu konfigurieren und zu erstellen. BigQuery bietet eine Query using Spark-Funktion, mit der Sie mit dem Codieren der Sitzung mit Vorlage beginnen können, wie auf dem Tab Spark-Sitzung mit Vorlage beschrieben.

Einzelne Sitzung

So erstellen Sie eine Spark-Sitzung in einem neuen Notebook:

1. Klicken Sie in der Tableiste des Editorbereichs auf den Drop-down-Pfeil neben dem +-Zeichen und dann auf Notebook.

   

2. Kopieren Sie den folgenden Code und führen Sie ihn in einer Notebook-Zelle aus, um eine einfache Spark-Sitzung zu konfigurieren und zu erstellen.

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()
)

Ersetzen Sie Folgendes:

• APP_NAME: Ein optionaler Name für Ihre Sitzung.
• Optionale Sitzungseinstellungen: Sie können Session-Einstellungen der Managed Service for Apache Spark API hinzufügen, um Ihre Sitzung anzupassen. Hier einige Beispiele:
  • RuntimeConfig:

    

    • 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.subnetwork_uri = "SUBNET_NAME"
    • session.environment_config.execution_config.ttl = {"seconds": VALUE}
    • session.environment_config.execution_config.service_account = SERVICE_ACCOUNT

Spark-Sitzung mit Vorlage

Sie können den Code in eine Notebook-Zelle eingeben und ausführen, um eine Spark-Sitzung basierend auf einer vorhandenen Sitzungsvorlage zu erstellen. Alle session-Konfigurationseinstellungen, die Sie in Ihrem Notebook-Code angeben, überschreiben die entsprechenden Einstellungen in der Sitzungsvorlage.

Wenn Sie schnell loslegen möchten, können Sie die Vorlage Query using Spark verwenden, um Ihr Notebook mit Spark-Sitzungsvorlagencode vorab auszufüllen:

1. Klicken Sie in der Tableiste des Editorbereichs auf den Drop-down-Pfeil neben dem +-Zeichen und dann auf Notebook.

   

2. Klicken Sie unter Mit einer Vorlage beginnen auf Abfrage mit Spark und dann auf Vorlage verwenden, um den Code in Ihr Notebook einzufügen.

   

3. Geben Sie die Variablen wie in den Hinweisen erläutert an.
4. Sie können alle zusätzlichen Beispielcodezellen löschen, die in das Notebook eingefügt wurden.

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()
)

Ersetzen Sie Folgendes:

• PROJECT_ID: Ihre Projekt-ID, die im Abschnitt Projektinformationen des Google Cloud Console-Dashboards aufgeführt ist.
• LOCATION: Die Compute Engine-Region, in der Ihre Notebook-Sitzung ausgeführt wird. Wenn nicht angegeben, wird die Region der VM verwendet, die das Notebook erstellt.

• SESSION_TEMPLATE: Der Name einer vorhandenen Vorlage für interaktive Sitzungen. Die Einstellungen für die Sitzungskonfiguration werden aus der Vorlage übernommen. In der Vorlage müssen außerdem die folgenden Einstellungen angegeben werden:

  • Laufzeitversion 2.3+

  • Notebook-Typ: Spark Connect

    Beispiel:

    

• APP_NAME: Ein optionaler Name für Ihre Sitzung.

PySpark-Code in Ihrem BigQuery Studio-Notebook schreiben und ausführen

Nachdem Sie eine Spark-Sitzung in Ihrem Notebook erstellt haben, können Sie damit Spark-Notebook-Code im Notebook ausführen.

Unterstützung der Spark Connect PySpark API: Ihre Spark Connect-Notebook-Sitzung unterstützt die meisten PySpark APIs, einschließlich DataFrame, Funktionen und Spalte, aber nicht SparkContext und RDD sowie andere PySpark APIs. Weitere Informationen finden Sie unter Was wird in Spark 3.5 unterstützt?.

Tipp: In der Spark SQL API-Referenz können Sie nachsehen, ob Spark Connect eine API unterstützt. Die Dokumentation für eine unterstützte API enthält die Meldung „Unterstützt Spark Connect“:

Direkte Schreibvorgänge in Spark Connect-Notebooks: Spark-Sitzungen in einem BigQuery Studio-Notebook konfigurieren den Spark BigQuery-Connector vor, um DIREKTE Datenschreibvorgänge auszuführen. Bei der DIRECT-Schreibmethode wird die BigQuery Storage Write API verwendet, die Daten direkt in BigQuery schreibt. Bei der INDIRECT-Schreibmethode, die die Standardmethode für Managed Service for Apache Spark-Batches ist, werden Daten in einen Zwischen-Cloud Storage-Bucket und dann in BigQuery geschrieben. Weitere Informationen zu INDIRECT-Schreibvorgängen finden Sie unter Daten aus BigQuery lesen und in BigQuery schreiben.

Managed Service for Apache Spark-spezifische APIs: Managed Service for Apache Spark vereinfacht das dynamische Hinzufügen von PyPI-Paketen zu Ihrer Spark-Sitzung, indem die Methode addArtifacts erweitert wird. Sie können die Liste im Format version-scheme angeben (ähnlich wie pip install). Dadurch wird der Spark Connect-Server angewiesen, Pakete und ihre Abhängigkeiten auf allen Clusterknoten zu installieren, sodass sie für Worker für Ihre UDFs verfügbar sind.

Beispiel, in dem die angegebene textdistance-Version und die neuesten kompatiblen random2-Bibliotheken auf dem Cluster installiert werden, damit UDFs mit textdistance und random2 auf Worker-Knoten ausgeführt werden können.

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

Notebook-Codehilfe: Das BigQuery Studio-Notebook bietet Codehilfe, wenn Sie den Mauszeiger auf einen Klassen- oder Methodennamen bewegen, und bietet Hilfe bei der Codevervollständigung, während Sie Code eingeben.

Wenn Sie im folgenden Beispiel DataprocSparkSession eingeben und den Mauszeiger auf diesen Klassennamen bewegen, werden die Codevervollständigung und die Dokumentationshilfe angezeigt.

Tipp: Informationen zur Verwendung von DataprocSparkSession.builder-Methoden zum Konfigurieren von Spark Connect-Sitzungen finden Sie unter Dataproc Spark Connect-Client auf GitHub.

PySpark-Beispiele für BigQuery Studio-Notebooks

In diesem Abschnitt finden Sie Beispiele für BigQuery Studio-Python-Notebooks mit PySpark-Code, um die folgenden Aufgaben auszuführen:

• Führen Sie eine Wortzählung für ein öffentliches Shakespeare-Dataset aus.
• Erstellen Sie eine Iceberg-Tabelle mit Metadaten, die im Lakehouse-Laufzeitkatalog gespeichert sind.

Wordcount

Im folgenden PySpark-Beispiel wird eine Spark-Sitzung erstellt und dann die Häufigkeit von Wörtern in einem öffentlichen bigquery-public-data.samples.shakespeare-Dataset gezählt.

# 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()

Ersetzen Sie Folgendes:

• APP_NAME: Ein optionaler Name für Ihre Sitzung.

Ausgabe:

In der Zellenausgabe wird ein Beispiel für die Wordcount-Ausgabe aufgeführt. Wenn Sie sich die Sitzungsdetails in der Google Cloud Console ansehen möchten, klicken Sie auf den Link Detailansicht der interaktiven Sitzung. Es wird ein Spark-Detailbereich mit Tabs für Logs, Executors, SQL und Jobs (einschließlich einer Ereigniszeitachse) geöffnet. Wenn Sie Ihre Spark-Sitzung überwachen möchten, klicken Sie auf der Seite mit den Sitzungsdetails auf Spark-UI ansehen.

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-Tabelle

PySpark-Code ausführen, um eine Iceberg-Tabelle mit Metadaten aus dem Lakehouse-Laufzeitkatalog zu erstellen

Im folgenden Beispielcode wird eine sample_iceberg_table mit Tabellenmetadaten erstellt, die im Lakehouse-Laufzeitkatalog gespeichert sind. Anschließend wird die Tabelle abgefragt.

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()

Hinweise:

• PROJECT_ID: Ihre Projekt-ID, die im Abschnitt Projektinformationen des Google Cloud Console-Dashboards aufgeführt ist.
• REGION und SUBNET_NAME: Geben Sie die Compute Engine-Region und den Namen eines Subnetzes in der Sitzungsregion an. Managed Service for Apache Spark aktiviert den privaten Google-Zugriff (PGA) für das angegebene Subnetz.
• LOCATION: Die Standardwerte für BigQuery_metastore_config.location und spark.sql.catalog.{catalog}.gcp_location sind US. Sie können jedoch einen beliebigen unterstützten BigQuery-Standort auswählen.
• BUCKET und WAREHOUSE_DIRECTORY: Der Cloud Storage-Bucket und der Ordner, die für das Iceberg-Warehouse-Verzeichnis verwendet werden.
• CATALOG und NAMESPACE: Der Name und der Namespace des Iceberg-Katalogs werden kombiniert, um die Iceberg-Tabelle (catalog.namespace.table_name) zu identifizieren.
• APP_NAME: Ein optionaler Name für Ihre Sitzung.

In der Zellenausgabe wird die sample_iceberg_table mit der hinzugefügten Spalte aufgeführt und ein Link zur Seite Details zur interaktiven Sitzung in der Google Cloud Console angezeigt. Es wird ein Spark-Detailbereich mit Tabs für Logs, Executors, SQL und Jobs (einschließlich einer Ereigniszeitachse) geöffnet. Sie können auf der Seite mit den Sitzungsdetails auf Spark-UI ansehen klicken, um Ihre Spark-Sitzung zu überwachen.

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)

Tabellendetails in BigQuery ansehen

Führen Sie die folgenden Schritte aus, um die Details einer Iceberg-Tabelle in BigQuery zu prüfen:

1. Rufen Sie in der Google Cloud Console die Seite BigQuery auf.

   BigQuery aufrufen

2. Klicken Sie im Bereich „Projektressourcen“ auf Ihr Projekt und dann auf Ihren Namespace, um die Tabelle sample_iceberg_table aufzulisten. Klicken Sie auf die Tabelle Details, um die Informationen zur Konfiguration für Katalogtabelle öffnen aufzurufen.

   Die Ein- und Ausgabeformate sind die Standard-Hadoop-Klassenformate InputFormat und OutputFormat, die von Iceberg verwendet werden.

   

Weitere Beispiele

Erstellen Sie einen Spark-DataFrame (sdf) aus einem Pandas-DataFrame (df).

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

Führen Sie Aggregationen in Spark DataFrames aus.

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()

Mit dem Spark-BigQuery-Connector aus BigQuery lesen.

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

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

Spark-Code mit Gemini Code Assist schreiben

Vorschau

Dieses Produkt oder diese Funktion unterliegt den „Pre-GA-Angebotsbedingungen“ im Abschnitt „Allgemeine Dienstbedingungen“ der dienstspezifischen Nutzungsbedingungen. Pre-GA-Produkte und -Features sind "im Ist-Zustand" verfügbar und bieten möglicherweise nur eingeschränkten Support. Weitere Informationen finden Sie unter den Beschreibungen der Startphase.

Sie können Gemini Code Assist auffordern, PySpark-Code in Ihrem Notebook zu generieren. Gemini Code Assist ruft relevante BigQuery- und Dataproc Metastore-Tabellen und deren Schemas ab und verwendet sie, um eine Codeantwort zu generieren.

So generieren Sie Gemini Code Assist-Code in Ihrem Notebook:

1. Fügen Sie eine neue Codezelle ein, indem Sie in der Symbolleiste auf + Code klicken. In der neuen Codezelle wird Start coding or generate with AI angezeigt. Klicken Sie auf Generieren.

2. Geben Sie im Editor „Generieren“ einen Prompt in natürlicher Sprache ein und klicken Sie dann auf enter. Achten Sie darauf, das Keyword spark oder pyspark in Ihren Prompt aufzunehmen.

   Beispiel-Prompt:

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

   Beispielausgabe:

   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")
   

Tipps zur Codegenerierung mit Gemini Code Assist

• Aktivieren Sie die Data Catalog-Synchronisierung für Dataproc Metastore-Instanzen, damit Gemini Code Assist relevante Tabellen und Schemas abrufen kann.

• Ihr Nutzerkonto muss Zugriff auf Data Catalog haben, um die Tabellen abzufragen. Weisen Sie dazu die Rolle DataCatalog.Viewer zu.

Spark-Sitzung beenden

Sie können eine der folgenden Aktionen ausführen, um Ihre Spark Connect-Sitzung in Ihrem BigQuery Studio-Notebook zu beenden:

• Führen Sie spark.stop() in einer Notebook-Zelle aus.
• Beenden Sie die Laufzeit im Notebook:
  1. Klicken Sie auf die Laufzeitauswahl und dann auf Sitzungen verwalten.

     

  2. Klicken Sie im Dialogfeld Aktive Sitzungen auf das Symbol zum Beenden und dann auf Beenden.

     

BigQuery Studio-Notebook-Code orchestrieren

Sie können BigQuery Studio-Notebookcode auf folgende Weise orchestrieren:

• Notebook-Code über die Google Cloud Console planen (es gelten die Notebook-Preise).

• Notebook-Code als Batch-Arbeitslast ausführen (es gelten die Preise für Managed Service for Apache Spark).

Notebook-Code über die Google Cloud Console planen

Sie können Notebook-Code auf folgende Weise planen:

• Planen Sie das Notebook.
• Wenn die Ausführung von Notebook-Code Teil eines Workflows ist, planen Sie das Notebook als Teil einer Pipeline.

Notebook-Code als Batcharbeitslast ausführen

Führen Sie die folgenden Schritte aus, um BigQuery Studio-Notebook-Code als Batcharbeitslast auszuführen.

1. Laden Sie den Notebook-Code in eine Datei in einem lokalen Terminal oder in Cloud Shell herunter.

   Es wird empfohlen, die Dateien in Cloud Shell herunterzuladen und dort zu bearbeiten, da dort Texteditoren und andere Tools vorinstalliert sind und Python-Unterstützung integriert ist.

   1. Öffnen Sie in der Google Cloud Console auf der Seite BigQuery Studio das Notebook im Bereich Explorer.

   2. Wenn Sie die Menüleiste maximieren möchten, klicken Sie auf keyboard_arrow_down Sichtbarkeit der Kopfzeile ein-/ausblenden.

   3. Klicken Sie auf Datei > Herunterladen und dann auf Download.py.

      

2. requirements.txt generieren.

   1. Installieren Sie pipreqs in dem Verzeichnis, in dem Sie die Datei .py gespeichert haben.

      pip install pipreqs
      

   2. Führen Sie pipreqs aus, um requirements.txt zu generieren.

      pipreqs filename.py
      

   3. Verwenden Sie die Google Cloud CLI, um die lokale Datei requirements.txt in einen Bucket in Cloud Storage zu kopieren.

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

3. Aktualisieren Sie den Spark-Sitzungscode, indem Sie die heruntergeladene Datei .py bearbeiten.

   1. Entfernen oder kommentieren Sie alle Shell-Scripts aus.

   2. Entfernen Sie den Code, der die Spark-Sitzung konfiguriert, und geben Sie dann Konfigurationsparameter als Parameter für die Übermittlung von Batcharbeitslasten an. (siehe Spark-Batcharbeitslast senden).

      Beispiel:

      • Entfernen Sie die folgende Konfigurationszeile für das Sitzungssubnetz aus dem Code:

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

      • Wenn Sie die Batcharbeitslast ausführen, verwenden Sie das Flag --subnet, um das Subnetz anzugeben.

        gcloud dataproc batches submit pyspark \
        --subnet=SUBNET_NAME
        

   3. Verwenden Sie ein einfaches Code-Snippet zum Erstellen einer Sitzung.

      • Beispiel für heruntergeladenen Notebook-Code vor der Vereinfachung.

        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()
        

      • Code der Batcharbeitslast nach der Vereinfachung.

        from pyspark.sql import SparkSession
        
        spark = SparkSession \
        .builder \
        .getOrCreate()
        

4. Führen Sie die Batcharbeitslast aus.

   1. Eine Anleitung finden Sie unter Spark-Batcharbeitslast senden.

      • Achten Sie darauf, das Flag „--deps-bucket“ anzugeben, um auf den Cloud Storage-Bucket zu verweisen, der Ihre requirements.txt-Datei enthält.

        Beispiel:

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

      Hinweise:

      • FILENAME: Der Name der heruntergeladenen und bearbeiteten Notebook-Codedatei.
      • REGION: Die Compute Engine-Region, in der sich Ihr Cluster befindet.
      • BUCKET: Der Name des Cloud Storage-Bucket, der Ihre requirements.txt-Datei enthält.
      • --version: Die Spark-Laufzeitversion 2.3 ist für die Ausführung der Batcharbeitslast ausgewählt.

5. Führen Sie einen Commit für Ihren Code durch.

   1. Nachdem Sie den Code für die Batcharbeitslast getestet haben, können Sie die Datei .ipynb oder .py mit Ihrem git-Client (z. B. GitHub, GitLab oder Bitbucket) als Teil Ihrer CI/CD-Pipeline in Ihr Repository übertragen.

6. Batch-Arbeitslast mit Managed Service for Apache Airflow planen

   1. Eine Anleitung finden Sie unter Managed Service for Apache Spark-Arbeitslasten mit Managed Airflow ausführen.

Notebook-Fehler beheben

Wenn in einer Zelle mit Spark-Code ein Fehler auftritt, können Sie den Fehler beheben, indem Sie in der Zellenausgabe auf den Link Detailansicht der interaktiven Sitzung klicken (siehe Beispiele für Wordcount und Iceberg-Tabelle).

Wenn ein Notebook-Codefehler auftritt, können Sie in der Spark-UI zum letzten Spark-Job navigieren, um zusätzliche Informationen zur Fehlerbehebung des fehlgeschlagenen Jobs zu erhalten.

Bekannte Probleme und Lösungen

Fehler: Eine Notebook-Laufzeit, die mit der Python-Version 3.10 erstellt wurde, kann einen PYTHON_VERSION_MISMATCH-Fehler verursachen, wenn versucht wird, eine Verbindung zur Spark-Sitzung herzustellen.

Lösung: Erstellen Sie die Laufzeit mit der Python-Version 3.11 neu.

Nächste Schritte

• YouTube-Videodemo: Unleashing the power of Apache Spark integrated with BigQuery.
• Lakehouse-Laufzeitkatalog mit Managed Service for Apache Spark verwenden
• Lakehouse-Laufzeitkatalog mit Managed Service for Apache Spark verwenden