{"filename":"snowflake-snowpark-dbt-cursorrules-prompt-file.mdc","content":"---\ndescription: \"Cursor rules for Snowpark Python (DataFrames, UDFs, stored procedures) and dbt with the Snowflake adapter.\"\nglobs: **/*\nalwaysApply: false\n---\n// Snowflake Snowpark Python & dbt\n// Expert guidance for Snowpark Python development and dbt with the Snowflake adapter\n\nYou are an expert in Snowpark Python (Snowflake's server-side Python API) and dbt with the dbt-snowflake adapter. You build production-grade data transformation pipelines using both tools.\n\n// ═══════════════════════════════════════════\n// SNOWPARK PYTHON\n// ═══════════════════════════════════════════\n\n// Snowpark runs Python server-side in Snowflake warehouses. Data never leaves Snowflake.\n// Core abstractions: Session, DataFrame, UDF, UDTF, UDAF, Stored Procedure.\n\n// Session\nfrom snowflake.snowpark import Session\nimport os\nsession = Session.builder.configs({\n    \"account\": os.environ[\"SNOWFLAKE_ACCOUNT\"],\n    \"user\": os.environ[\"SNOWFLAKE_USER\"],\n    \"password\": os.environ[\"SNOWFLAKE_PASSWORD\"],\n    \"role\": \"my_role\", \"warehouse\": \"my_wh\", \"database\": \"my_db\", \"schema\": \"my_schema\"\n}).create()\n\n// DataFrame API — Lazy evaluation, builds query plan executed on collect()/show().\ndf = session.table(\"customers\")\ndf_filtered = df.filter(df[\"region\"] == \"US\").select(\"name\", \"email\", \"revenue\")\ndf_agg = df.group_by(\"region\").agg(sum(\"revenue\").alias(\"total_revenue\"))\ndf_agg.show()\n\n// Key operations: .filter(), .select(), .group_by().agg(), .join(), .sort(),\n// .with_column(), .drop(), .distinct(), .limit(), .union_all(), .flatten(),\n// .write.save_as_table()\n\n// Scalar UDFs\nfrom snowflake.snowpark.functions import udf\n@udf(name=\"normalize_email\", replace=True)\ndef normalize_email(email: str) -> str:\n    return email.strip().lower() if email else None\n\n// Vectorized UDFs (10-100x faster for ML inference):\nimport pandas as pd\n@udf(name=\"predict_score\", packages=[\"scikit-learn\", \"pandas\"], replace=True)\ndef predict_score(features: pd.Series) -> pd.Series:\n    import pickle, sys\n    model = pickle.load(open(sys.path[0] + \"/model.pkl\", \"rb\"))\n    return pd.Series(model.predict(features.values.reshape(-1, 1)))\n\n// UDTFs (return multiple rows per input):\nclass Tokenizer:\n    def process(self, text: str):\n        for token in text.split():\n            yield (token,)\n\ntokenize = session.udtf.register(Tokenizer,\n    output_schema=StructType([StructField(\"token\", StringType())]),\n    input_types=[StringType()], name=\"tokenize\", replace=True)\n\n// Stored Procedures (server-side multi-step logic):\nfrom snowflake.snowpark.functions import sproc\n@sproc(name=\"daily_etl\", replace=True, packages=[\"snowflake-snowpark-python\"])\ndef daily_etl(session: Session) -> str:\n    raw = session.table(\"raw_events\")\n    cleaned = raw.filter(raw[\"event_type\"].is_not_null())\n    cleaned.write.mode(\"overwrite\").save_as_table(\"cleaned_events\")\n    return f\"Processed {cleaned.count()} rows\"\n\n// Third-Party Packages: session.add_packages(\"pandas\", \"scikit-learn==1.3.0\", \"xgboost\")\n// File Access: session.add_import(\"@my_stage/model.pkl\") for static files.\n// pandas on Snowflake (no data movement):\n//   import modin.pandas as pd; import snowflake.snowpark.modin.plugin\n//   df = pd.read_snowflake(\"my_table\")\n\n// ═══════════════════════════════════════════\n// DBT WITH SNOWFLAKE ADAPTER\n// ═══════════════════════════════════════════\n\n// Install: pip install dbt-snowflake\n// profiles.yml:\nmy_project:\n  target: dev\n  outputs:\n    dev:\n      type: snowflake\n      account: myaccount\n      user: myuser\n      password: \"{{ env_var('SNOWFLAKE_PASSWORD') }}\"\n      role: transformer\n      database: analytics\n      warehouse: transforming\n      schema: public\n      threads: 4\n\n// Materializations: view, table, incremental, ephemeral, dynamic_table\n\n// Dynamic Tables in dbt:\n// {{ config(materialized='dynamic_table', snowflake_warehouse='transforming', target_lag='1 hour') }}\n// SELECT customer_id, SUM(amount) AS lifetime_value FROM {{ ref('stg_orders') }} GROUP BY 1\n\n// Incremental Models:\n{{\n  config(\n    materialized='incremental',\n    unique_key='event_id',\n    incremental_strategy='merge',\n    on_schema_change='sync_all_columns'\n  )\n}}\nSELECT * FROM {{ ref('stg_events') }}\n{% if is_incremental() %}\n  WHERE event_timestamp > (SELECT MAX(event_timestamp) FROM {{ this }})\n{% endif %}\n\n// Snowflake-Specific Configs:\n// cluster_by=['col1', 'col2']  — Clustering (large tables only)\n// transient=true               — No Fail-safe (lower storage cost)\n// query_tag='finance_daily'    — Workload attribution\n// copy_grants=true             — Preserve access on replace\n// snowflake_warehouse='lg_wh'  — Per-model warehouse override\n// secure=true                  — Secure views\n\n// Sources (models/staging/_sources.yml):\nsources:\n  - name: raw\n    database: raw_db\n    schema: jaffle_shop\n    tables:\n      - name: customers\n        loaded_at_field: _loaded_at\n        freshness:\n          warn_after: {count: 12, period: hour}\n          error_after: {count: 24, period: hour}\n\n// Testing (schema.yml):\nmodels:\n  - name: stg_customers\n    columns:\n      - name: customer_id\n        tests: [unique, not_null]\n\n// Key Commands:\n// dbt run, dbt test, dbt build (run+test in order), dbt compile\n// dbt run --select my_model+  (model + downstream)\n// dbt run --select +my_model  (model + upstream)\n// dbt source freshness, dbt docs generate && dbt docs serve\n\n// Custom schema macro (macros/generate_schema_name.sql):\n{% macro generate_schema_name(custom_schema_name, node) %}\n  {% if custom_schema_name %}{{ custom_schema_name | trim }}{% else %}{{ target.schema }}{% endif %}\n{% endmacro %}\n\n// Best Practices\n- Prefer vectorized UDFs (pandas) for ML inference — much faster than scalar UDFs.\n- Pin package versions in production UDFs and stored procedures.\n- Use DataFrame API over raw SQL strings in reusable Python pipelines.\n- Use staging models (stg_*) to rename/type-cast, mart models for business tables.\n- Use incremental for fact tables; dynamic_table for near-real-time.\n- Set on_schema_change='sync_all_columns' on incremental models.\n- Use copy_grants=true to avoid permission issues. Tag models for selective execution.\n- Use separate warehouses for dbt runs vs analyst queries.\n\n// Anti-Patterns\n- Do NOT collect() large DataFrames to client — process server-side.\n- Do NOT use Python loops over rows — use DataFrame operations or vectorized UDFs.\n- Do NOT use {{ this }} without {% if is_incremental() %} guard.\n- Do NOT set cluster_by on small tables (< 1TB).\n- Do NOT use materialized='table' for everything — views are free.\n- Do NOT hardcode database/schema — use {{ ref() }} and {{ source() }}.\n"}