Cursor Snowflake 编码规范：来自 PatrickJS/awesome-cursorrules (40k stars) 的 sno，适用于各类文档与内容的智能化处理。
Cursor Snowflake 编码规范：来自 PatrickJS/awesome-cursorrules (40k stars) 的 sno，适用于各类文档与内容的智能化处理。
**文件来源:** PatrickJS/awesome-cursorrules → `rules/snowflake-snowpark-dbt-cursorrules-prompt-file.mdc`
**原仓库:** https://github.com/PatrickJS/awesome-cursorrules
**评分:** ⭐ 仓库 40k stars (社区最权威 Cursor rules 合集)
把 `snowflake-snowpark-dbt-cursorrules-prompt-file.mdc` 这条 Cursor 编码规则打包成可调用的 AI skill,帮你把代码生成统一到一致的标准上。
> Cursor rules for Snowpark Python (DataFrames, UDFs, stored procedures) and dbt with the Snowflake adapter.
globs: **/*
alwaysApply: false
---
// Snowflake Snowpark Python & dbt
// Expert guidance for Snowpark Python development and dbt with the Snowflake adapter
You 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.
// ═══════════════════════════════════════════
// SNOWPARK PYTHON
// ═══════════════════════════════════════════
// Snowpark runs Python server-side in Snowflake warehouses. Data never leaves Snowflake.
// Core abstractions: Session, DataFrame, UDF, UDTF, UDAF, Stored Procedure.
// Session
from snowflake.snowpark import Session
import os
session = Session.builder.configs({
"account": os.environ["SNOWFLAKE_ACCOUNT"],
"user": os.environ["SNOWFLAKE_USER"],
"password": os.environ["SNOWFLAKE_PASSWORD"],
"role": "my_role", "warehouse": "my_wh", "database": "my_db", "schema": "my_schema"
}).create()
// DataFrame API — Lazy evaluation, builds query plan executed on collect()/show().
df = session.table("customers")
df_filtered = df.filter(df["region"] == "US").select("name", "email", "revenue")
df_agg = df.group_by("region").agg(sum("revenue").alias("total_revenue"))
df_agg.show()
// Key operations: .filter(), .select(), .group_by().agg(), .join(), .sort(),
// .with_column(), .drop(), .distinct(), .limit(), .union_all(), .flatten(),
// .write.save_as_table()
// Scalar UDFs
from snowflake.snowpark.functions import udf
@udf(name="normalize_email", replace=True)
def normalize_email(email: str) -> str:
return email.strip().lower() if email else None
// Vectorized UDFs (10-100x faster for ML inference):
import pandas as pd
@udf(name="predict_score", packages=["scikit-learn", "pandas"], replace=True)
def predict_score(features: pd.Series) -> pd.Series:
import pickle, sys
model = pickle.load(open(sys.path[0] + "/model.pkl", "rb"))
return pd.Series(model.predict(features.values.reshape(-1, 1)))
// UDTFs (return multiple rows per input):
class Tokenizer:
def process(self, text: str):
for token in text.split():
yield (token,)
tokenize = session.udtf.register(Tokenizer,
output_schema=StructType([StructField("token", StringType())]),
input_types=[StringType()], name="tokenize", replace=True)
// Stored Procedures (server-side multi-step logic):
from snowflake.snowpark.functions import sproc
@sproc(name="daily_etl", replace=True, packages=["snowflake-snowpark-python"])
def daily_etl(session: Session) -> str:
raw = session.table("raw_events")
cleaned = raw.filter(raw["event_type"].is_not_null())
cleaned.write.mode("overwrite").save_as_table("cleaned_events")
return f"Processed {cleaned.count()} rows"
// Third-Party Packages: session.add_packages("pandas", "scikit-learn==1.3.0", "xgboost")
// File Access: session.add_import("@my_stage/model.pkl") for static files.
// pandas on Snowflake (no data movement):
// import modin.pandas as pd; impo
...(完整内容在原仓库)...有问题或建议,在本 skill 下留言。
把 `snowflake-snowpark-dbt-cursorrules-prompt-file.mdc` 这条 Cursor 编码规则打包成可调用的 AI skill,帮你把代码生成统一到一致的标准上。
> Cursor rules for Snowpark Python (DataFrames, UDFs, stored procedures) and dbt with the Snowflake adapter.