Managed Spark on Google Cloud

[!IMPORTANT]

You MUST follow the Task Execution Workflow when writing spark code.

Notebook Generation Environment Rules

[!CAUTION]

When the task is to generate a Spark notebook (as opposed to executing one), you MUST NOT modify your own local execution environment (i.e. do not run these commands with your shell or other tools). Specifically:

• NEVER create a virtual environment (python -m venv, virtualenv, conda create, uv venv, poetry init/install, etc.).
• NEVER install packages (pip install, %pip install, uv pip install, conda install, poetry add).
• NEVER run environment discovery loops such as activating a venv, pip list, or probing for ipykernel / google-cloud-spark-connect as a precondition for writing the notebook.

Assume the user will select the appropriate kernel and dependencies after the notebook is generated.

Task Execution Workflow

1. Understand user request:

   • When asked to generate a new spark notebook you MUST clarify with the user which kernel type will be used: "Local Python" (Spark Connect) or "Remote Spark" where the Spark session (spark variable) is automatically initialized. If the user selects "Local Python", you MUST additionally clarify which dataprocSessionConfig should be used (scan for existing serverless session templates), and MUST initialize the session with ManagedSparkSession by adding the following cells to the notebook:

     from google.cloud.dataproc_v1 import Session
     # requires google-cloud-spark-connect pypi package for execution
     from google.cloud.managed_spark_connect import ManagedSparkSession
     
     session_config = Session()
     session_config.session_template = (
         "projects/<PROJECT_ID>/locations/<REGION>/sessionTemplates/<TEMPLATE_ID>"
     )
     spark = (
         ManagedSparkSession.builder.projectId("<PROJECT_ID>")
         .location("<REGION>")
         .dataprocSessionConfig(session_config)
         .getOrCreate()
     )

2. Understand schemas: ALWAYS use @skill:discovering-gcp-data-assets skill or references/schema_direct_inspection.md to understand input and output schemas. Include the schema in your thought process BEFORE generating any code. Do NOT guess column names. Cap GCP data asset discovery attempts at 3 retries max. Unless explicitly specified, assume that the assets are located in the same project. Avoid scanning for assets across other projects as it can take a long time. If an expected dataset or table does not exist, use @skill:discovering-gcp-data-assets to discover all similar tables in the namespace or project.

   MISSING SOURCE RULE: If the request does not name a table, GCS path or catalog, you MUST run at most ONE @skill:discovering-gcp-data-assets search. If it does not return a single clear match, you MUST list the candidates, ask the user which source to use, and STOP. You MUST NOT infer the source by listing buckets, reading existing notebooks or scripts, or running SHOW NAMESPACES / SHOW TABLES.

   MINOR TYPO RULE: If there is a minor typo (e.g. employees vs employee), you can fix the error and proceed.

   STRICT HALT RULE: If the discovered table names differ from the requested table by more than a minor typo (e.g. completely different words, prefixes, or suffixes), you must IMMEDIATELY report the missing table and a neutral list of all available alternatives in the same namespace to the user without making any recommendations. You MUST ask the user which alternative to use and then STOP EXECUTING your turn. Do NOT write any Spark code or notebooks. Do NOT proceed with code generation, do NOT add fallback logic to code, and do NOT automatically substitute any alternative table (even if its schema seems to match) without explicit user permission.

3. Verify source accessibility: For Google Cloud Storage (gs://) paths, verify access/existence using gcloud storage ls gs://<path-to-dataset> (skip gcloud storage ls for external cloud URIs such as s3a://, abfss://, wasbs://, or wasb://, which are accessed directly by Spark's Hadoop FileSystem connectors). If accessing or reading a GCS path fails with a storage error e.g., permission errors like 403 Forbidden/Forbidden/PermissionDenied, or location errors like 404 Not Found/NotFound/FileNotFoundException you should report the error immediately. Either (1) ask the user what to do next, or (2) if asked to execute a notebook, save the notebook with the error output and recommend next steps to resolve the issue. Do NOT scan all buckets for alternative fallback datasets when encountering GCS errors.

4. Generate spark code:

   • Output Format: ALWAYS generate code in Python Notebooks (.ipynb) format. Generate scripts (.py) only if explicitly requested.

   • Spark Session Initialization:

     [!IMPORTANT] Initializing a Spark session on Google Cloud can take 2-3 minutes. You MUST inform the user about the potential delay.

     [!CAUTION] You MUST NOT create a local Spark session. The following are BANNED: - SparkSession.builder.master("local") - pyspark.sql.SparkSession.builder.getOrCreate() in notebooks or local Spark Connect scripts - Any try/except fallbacks that revert to a local SparkSession.

     • Remote Kernel Notebooks (.ipynb): spark is pre-initialized; use it directly without calling ManagedSparkSession or SparkSession.builder.getOrCreate().
     • Local Python Notebooks (.ipynb) & Spark Connect scripts: You MUST use ManagedSparkSession from google-cloud-spark-connect to connect to Managed Spark Serverless.
     • Dataproc Batch & Cluster jobs (.py submitted via gcloud dataproc batches submit pyspark or gcloud dataproc jobs submit pyspark): Use standard SparkSession.builder.appName("<APP_NAME>").getOrCreate() (never .master("local")).

     Refer to references/gcloud_dataproc.md for detailed configuration. Minimal initialization for Local Python notebooks / Spark Connect:

     from google.cloud.managed_spark_connect import ManagedSparkSession
     
     spark = (
         ManagedSparkSession.builder.projectId("<PROJECT_ID>")
         .location("<REGION>")
         .getOrCreate()
     )

   • Production Logging: In standalone production PySpark batch jobs and scripts (.py), you MUST use the standard Python logging module instead of print() statements for job lifecycle, progress, and record counts. Do NOT use the logging module in notebooks (.ipynb); use print(), .show(), and standard cell outputs instead. Configure logging for .py scripts with timestamps:

     import logging
     
     logging.basicConfig(
         level=logging.INFO,
         format="%(asctime)s [%(levelname)s] %(message)s",
     )

   • Security & Dynamic Variables: You MUST NOT hardcode plaintext credentials, passwords, or tokens (use Secret Manager — refer to references/jdbc_cloudsql.md when retrieving credentials from Secret Manager). You MUST NOT hardcode environment-specific variables; parameterize project IDs, dataset IDs, table names, and bucket paths via os.environ.get(...) or arguments.
   • Read and Write data: ALWAYS load only the connector reference file(s) needed for the prompt:
     • BigQuery, GCS, or Pub/Sub: references/read_write_data.md
     • BigLake Iceberg Catalog: references/biglake_iceberg.md
     • Cloud Spanner: references/spanner.md
     • JDBC, Cloud SQL (PostgreSQL / MySQL / SQL Server), AlloyDB & Secret Manager: references/jdbc_cloudsql.md
     • Amazon S3 (s3a:// / S3-compatible object storage): references/aws_s3.md
     • Azure Blob Storage & ADLS Gen2 (abfss://, wasbs://, wasb://): references/azure_storage.md
   • Machine Learning Tasks: Refer to @skill:ml-best-practices skill and references/ml_tasks.md when generating Machine Learning code.
   • Spark Optimizations & Broadcast Joins: ALWAYS refer to references/spark_optimizations.md. When joining a large DataFrame with a small lookup or dimension table, you MUST use broadcast() (from pyspark.sql.functions import broadcast) on the small table. Treat about 10 MB (Spark's default spark.sql.autoBroadcastJoinThreshold) as a guide to what counts as small, not a hard limit.
5. Verify schema before write: ALWAYS verify that the dataframe and destination schema match, use df.printSchema() for dataframe schema and refer to @skill:discovering-gcp-data-assets skill or references/schema_direct_inspection.md to verify destination schema.
6. Compile code before executing: For notebooks convert them to python script using jupyter nbconvert --to script your-notebook.ipynb first. Then compile the resulting python script using python3 -m py_compile your-script.py. The same can be done for pyspark source code.
7. Execute script or notebook: When requested to run a job, script, session, or execute notebook cells against Managed Spark, refer to references/gcloud_dataproc.md on how to execute code on Dataproc Serverless using Spark Connect or Dataproc jobs.
8. Follow up with the user: If a brand new notebook was generated, instruct the user to select the appropriate kernel in the dropdown for cell execution.

Common Mistakes Checklist

[!CAUTION]

Ensure you verify this checklist to avoid mistakes

Before submitting a job, verify:

• All imports present (col, when, lit, broadcast, etc. from pyspark.sql.functions)
• vector_to_array from correct module use from pyspark.ml.functions import vector_to_array (NOT pyspark.sql.functions)
• DataFrame schema matches target Iceberg table verify with df.printSchema() before writing
• CSV files read with header and inferSchema without these, the header row becomes data and all columns are strings
• Driver memory safety (toPandas() / collect()) NEVER call .toPandas() or .collect() on raw or un-aggregated DataFrames. ALWAYS perform transformations, aggregations (groupBy().agg()), or data reduction (limit(), sample()) in Spark before converting small summaries to Pandas for plotting or display.
• No inline pip install in Spark batch/script jobs: NEVER run pip install or subprocess package installations inside PySpark batch scripts (the %pip install setup cell in generated notebooks is exempt). Pass dependencies using --properties=spark.jars.packages=..., --archives=gs://.../env.tar.gz#environment, --py-files, or a custom --container-image.
• Optimization & Pre-Submission Refactoring Verify the job against references/spark_optimizations.md and follow the confirmation protocol in references/spark_refactoring_guide.md.
• No full file reads of executed notebooks: You MUST NOT use read_file or generic full file reading tools on executed .ipynb notebooks; they often contain massive base64-encoded image outputs that cause excessive token consumption. You MUST inspect execution outputs with cell-scoped reading tools (such as notebook__read_cell, jupyter__read_cell, or specific line slices).

IAM Requirements

The Managed Spark (Dataproc) service account needs:

• roles/dataproc.worker: Job execution
• roles/biglake.admin: BigLake Iceberg table management
• roles/bigquery.jobUser: BigQuery query materialization and load jobs
• roles/bigquery.dataEditor: BigQuery table reads and writes (direct and indirect)
• roles/bigquery.readSessionUser: BigQuery Storage Read API session creation
• roles/storage.objectUser: Read/write GCS (data, Iceberg warehouse, and --deps-bucket / temporaryGcsBucket)
• roles/spanner.databaseUser: Spanner reads and writes
• roles/secretmanager.secretAccessor: Accessing database passwords and credentials stored in Secret Manager

Spark resource management & observability

Refer to references/gcloud_dataproc.md for detailed guidelines on managing Spark clusters, jobs, batches, interactive sessions, and Spark Connect sessions.

For fleet-wide and multi-region observability — including active workload inventory across Batches, Sessions, Clusters, and Jobs, failed batch discovery, Serverless compute consumption (DCU-Hours and Shuffle-GB-Months), idle or zombie cluster waste detection, scheduled pipeline cadence health, configuration auditing, and Cloud Monitoring / Cloud Logging telemetry — use the @skill:gcp-spark-observability skill.

Code Optimization & Pre-Submission Verification

Before submitting any Spark job or Dataproc batch, enforce the pre-submission verification protocol in references/spark_refactoring_guide.md: inspect the code for the anti-patterns catalogued in references/spark_optimizations.md and obtain explicit user confirmation before applying refactoring.

Troubleshooting & Root Cause Analysis

For troubleshooting failed Spark jobs or Dataproc batches, inspecting/tailing GCS driver output logs, analyzing Spark event logs, or performing Root Cause Analysis (RCA), use the @skill:gcp-spark-troubleshooting skill.