重构Python代码适配Snowpark（一）：基础代码运行问题求助

原Python代码 #

x_pos = 0

def move(x_pos, by_amount=9998):
    x_pos += by_amount
    return x_pos

final_x_pos = move(x_pos, 13)
print(final_x_pos)

三次失败尝试及错误信息 #

尝试一 #

x_pos = 0

def main(session: snowpark.Session, x_pos, by_amount=9998):
    x_pos += by_amount
    return x_pos

final_x_pos = main(x_pos, 13)
print(final_x_pos)

错误：Handler has more arguments than expected. Function signature must have exactly one argument:
def main(session: snowpark.Session):

尝试二 #

x_pos = 0

def move(x_pos, by_amount=9998):
    x_pos += by_amount
    return x_pos

final_x_pos = move(x_pos, 13)
print(final_x_pos)
def main(session: snowpark.Session, x_pos, by_amount):
    return(final_x_pos)

错误：Handler has more arguments than expected. Function signature must have exactly one argument:
def main(session: snowpark.Session):

尝试三 #

def add_num(num1,num2):
    return num1+num2
    result = add_num(10,20)
def main(session: snowpark.Session):
    return (add_num)

注：此代码无明确报错，但存在逻辑错误——result赋值语句在return之后永远不会执行，且main返回函数对象而非计算结果，不符合Snowpark执行要求。

核心问题原因 #

Snowpark的入口Handler函数（即main函数）强制要求仅接受snowpark.Session这一个参数，不能额外添加自定义参数；同时main需要返回可被Snowpark识别的结果（如数据帧、标量值等），而非函数对象。

解决方案 #

方案一：本地调试适配 #

将原有业务逻辑全部放入main函数内部执行，确保main仅接收Session参数：

from snowflake.snowpark import Session

def move(x_pos, by_amount=9998):
    x_pos += by_amount
    return x_pos

def main(session: Session):
    # 原代码逻辑移至main内部
    x_pos = 0
    final_x_pos = move(x_pos, 13)
    print(final_x_pos)
    return final_x_pos

# 本地测试时创建Snowpark会话
if __name__ == "__main__":
    # 替换为你的Snowflake连接配置
    connection_config = {
        "account": "你的账户名",
        "user": "你的用户名",
        "password": "你的密码",
        "warehouse": "你的仓库",
        "database": "你的数据库",
        "schema": "你的模式",
        "role": "你的角色"
    }
    session = Session.builder.configs(connection_config).create()
    main(session)

方案二：封装为Snowpark UDF（云环境运行） #

如果需要在Snowflake云端执行该逻辑，将业务函数注册为用户自定义函数（UDF）：

from snowflake.snowpark import Session
from snowflake.snowpark.functions import udf

def move(x_pos, by_amount=9998):
    x_pos += by_amount
    return x_pos

def main(session: Session):
    # 注册UDF到Snowflake
    move_udf = udf(move, return_type=int, input_types=[int, int])
    
    # 示例：用DataFrame调用UDF
    test_data = session.create_dataframe([(0, 13)], schema=["初始位置", "移动量"])
    result_df = test_data.with_column("最终位置", move_udf(test_data["初始位置"], test_data["移动量"]))
    
    # 打印结果
    result_df.show()
    return result_df.collect()

# 本地测试
if __name__ == "__main__":
    connection_config = {
        # 替换为你的Snowflake连接参数
    }
    session = Session.builder.configs(connection_config).create()
    main(session)

关键注意点 #

1. 所有外部参数或变量需放入main内部，或通过Session传递（如读取Snowflake表数据）；
2. 若要在云端运行，必须通过UDF、存储过程等Snowpark支持的方式封装业务逻辑；
3. 本地测试时需正确配置Snowflake会话连接参数，确保能正常连接到你的Snowflake账户。

内容的提问来源于stack exchange，提问作者Patterson