Validate Pyspark To Snowpark Connect

1. Never modify the migrated workload. Create an editable copy in <workload>_test/ and add a single entrypoint.py file that triggers the main execution flow.
2. Create synthetic data for all external sources. Mock Category A sources (spark.table()) by checking if tables exist and creating them if permitted. Mock Category B sources (spark.read.* with cloud/local paths) by uploading synthetic files to a Snowflake stage and replacing paths in the workload copy (see Phases 1.3 and 2.1).
3. Minimal synthetic data. 2-5 rows per source. Only include columns actually used by the workload.
4. CRITICAL: Import and call the REAL workload functions. The entrypoint must import the actual migrated code (e.g., from modeling_library import model, load_data) and call it. Do NOT rewrite or duplicate workload logic — no independent test cases like "Test window functions" or "Test joins".
5. This is a smoke test, not a unit test suite. The goal is to verify the workload runs end-to-end without exceptions, not to test individual operations or assert data correctness.

Prerequisites

uv --version || echo "PREREQ_FAIL: uv not installed"
uv run --project <SKILL_DIRECTORY> \
  python -c "from snowflake import snowpark_connect; spark = snowpark_connect.init_spark_session(); print('OK')" \
  || echo "PREREQ_FAIL: Snowflake connection failed"

# Check 3 (notebook workloads only): jupyter nbconvert
uv run --project <SKILL_DIRECTORY> \
  jupyter nbconvert --version \
  || echo "PREREQ_FAIL: jupyter nbconvert not installed"

Workflow

You MUST perform the phases below in order.

Phase 1: Analyze Workload

1.1 Validate migrated workload exists

test -e "$ARGUMENTS" || echo "ABORT: Migrated workload not found"

1.2 Identify and classify external data dependencies

Find all external data access in the workload: spark.read.*, spark.table(), spark.sql("SELECT ... FROM ..."), boto3/S3.

For directory workloads, scan ALL .py files in the directory — not just the main entrypoint. Data reads may occur in any module (e.g., a loader.py or data_access.py).

For single .py files, search the source directly. For .ipynb notebooks, search within the source arrays of code cells in the notebook JSON.

Classify each data source into one of two categories:

Category A — Table-mockable (handled via Snowflake tables):

• spark.table("table_name") calls
• spark.sql("SELECT ... FROM table_name") calls

For each, determine:

• Fully qualified table name (e.g., database.schema.table_name)
• Column names and types (infer from downstream usage)

Category B — Stage-mockable (requires Snowflake stage with synthetic files):

• spark.read.csv("s3://..."), spark.read.parquet("gs://..."), spark.read.json("/mnt/..."), or any spark.read.* call with an external cloud path (s3://, s3a://, gs://, abfs://, wasb://, adl://) or local/mounted path
• Also check for spark.read.format(...).load("path") variants

For each file read, capture:

• Read method: csv, parquet, json, text, or format(...).load(...)
• Full path string: e.g., s3://analytics-lake/raw/events/2024/
• Read options: e.g., header=True, inferSchema=True, delimiter=","
• Downstream column usage: trace how the resulting DataFrame is used (selects, joins, filters, casts) to infer the schema — same approach as Phase 2.3

1.2.1 Stopping Point — External File Reads

If any Category B (stage-mockable) reads were found, you MUST pause and alert the user:

⚠️  External cloud reads detected in the migrated workload:
  1. spark.read.<method>("<path>")
  2. ...

These paths reference external cloud storage (S3, GCS, Azure Blob, etc.).
Snowflake recommends creating an external stage that points to these
cloud locations for production use.

For validation, I need a Snowflake stage to upload synthetic test data.

Do you already have an external stage for these locations?
  - If YES: provide the stage name and I'll upload mock data files to it.
  - If NO: I'll create an internal stage (SCOS_VALIDATION_<workload>)
    and upload synthetic files there for testing.

Wait for the user to respond before proceeding.

• User provides an existing stage name → Record it as <STAGE_NAME>. Skip stage creation in Phase 1.3.1 (stage already exists). Proceed to Phase 1.3.2 to generate and upload mock data.
• User says no / no existing stage → Use the auto-generated name SCOS_VALIDATION_<workload> as <STAGE_NAME>. Create it as an internal stage in Phase 1.3.1.
• User provides a custom name for a new stage → Use that name as <STAGE_NAME>. Create it in Phase 1.3.1.

Use a single stage for ALL mock files. Do NOT create multiple stages — store all synthetic data files in one stage, organized by subpath (e.g., @<STAGE_NAME>/events/, @<STAGE_NAME>/users/).

1.2.2 Stopping Point — Table Dependencies

If any Category A (table-mockable) reads were found, you MUST pause and verify table existence in Snowflake before proceeding:

1. Check if the tables exist using Snowflake SHOW TABLES or SELECT COUNT(*).

uv run --project <SKILL_DIRECTORY> python -c "
from snowflake.snowpark import Session
session = Session.builder.config('connection_name', 'default').create()