使用 Python 编写存储过程¶

限制¶

Snowpark Stored Procedures 具有以下限制：

• 存储过程不支持创建进程。

• 存储过程不支持运行并发查询。

• 您不能使用执行 PUT 和 GET 命令的 APIs，包括 Session.sql("PUT ...") 和 Session.sql("GET ...")。

• 当您使用 session.file.get 从暂存区下载文件时，不支持模式匹配。

• 如果从任务执行存储过程，则必须在创建任务时指定仓库。您不能使用无服务器计算资源来运行任务。

• 所有者权限存储过程不支持创建指定的临时对象。所有者权限存储过程是以存储过程所有者的权限运行的存储过程。有关更多信息，请参阅 :doc:` 调用方的权限或所有者的权限 </developer-guide/stored-procedure/stored-procedures-rights>`。

计划编写存储过程¶

存储过程在 Snowflake 内部运行，因此您在规划代码时必须考虑到这一点。

• 限制消耗的内存量。Snowflake 根据所需的内存量对方法进行了限制。如需指导，请参阅 设计保持在 Snowflake 施加的约束范围内的处理程序。

• 确保您的处理程序方法或函数是线程安全的。

• 遵守规则和安全限制。请参阅 UDFs 和过程的安全实践。

• 确定是否希望以 :doc:` 调用者权限或所有者权限 </developer-guide/stored-procedure/stored-procedures-rights>` 运行存储过程。

• 考虑用于运行存储过程的 snowflake-snowpark-python 版本。由于存储过程发布流程的限制，Python 存储过程环境中可用的 snowflake-snowpark-python 库通常比公开发布的版本落后一个版本。使用以下 SQL 查找最新可用版本：

  select * from information_schema.packages where package_name = 'snowflake-snowpark-python' order by version desc;
  

  Copy

编写方法或函数¶

为存储过程编写方法或函数时，请注意以下事项：

• 指定 Snowpark Session 对象作为方法或函数的第一个实参。调用存储过程时，Snowflake 会自动创建一个 Session 对象并将其传递给存储过程。（您不能自行创建 Session 对象。）

• 对于其余实参和返回值，请使用与 Snowflake 数据类型 中列出的 Python 数据类型。

• 在过程的处理程序中运行异步子作业（例如使用 DataFrame .collect_nowait）时，不支持 “发后即忘”。

  换句话说，如果处理程序发出的子查询在父存储过程作业完成时仍在运行，子作业就会自动取消。

处理错误¶

您可以使用正常的 Python 异常处理技术来捕获过程中的错误。

如果方法内部出现未捕获的异常，Snowflake 会引发一个包含异常堆栈跟踪的错误。启用 记录未处理异常 后，Snowflake 会在事件表中记录有关未处理异常的数据。

为代码提供依赖项¶

如果处理程序代码依赖在处理程序之外（如模块中定义的代码）定义的代码或依赖资源文件，则可以通过将这些依赖项上传到暂存区来为代码提供这些依赖项。请参阅 为代码提供依赖项，或者对于 Python 工作表，请参阅 将 Python 文件从暂存区添加到工作表。

如果使用 SQL 创建存储过程，请在编写 CREATE PROCEDURE 语句 时使用 IMPORTS 子句，以指向依赖项文件。

通过存储过程访问 Snowflake 中的数据¶

要访问 Snowflake 中的数据，请使用 Snowpark 库 APIs。

处理对 Python 存储过程的调用时，Snowflake 会创建一个 Snowpark Session 对象，并将该对象传递给存储过程的方法或函数。

与使用其他语言的存储过程一样，会话的上下文（例如权限、当前数据库和架构等）取决于存储过程是以调用方权限还是所有者权限运行。有关详细信息，请参阅 访问和设置会话状态。

您可以使用此 Session 对象调用 Snowpark 库 中的 APIs。例如，您可以 :ref:` 为表创建 DataFrame <label-snowpark_python_dataframes>`，或执行 SQL 语句。

有关更多信息，请参阅 :doc:` Snowpark 开发者指南 </developer-guide/snowpark/python/index>`。

def run(session, from_table, to_table, count):

  session.table(from_table).limit(count).write.save_as_table(to_table)

  return "SUCCESS"

读取文件¶

您可以使用 Snowpark snowflake.snowpark.files 模块中的 SnowflakeFile 类从 Python 处理程序中的暂存区动态读取文件。

Snowflake 支持使用 SnowflakeFile 读取存储过程和用户定义函数的文件。有关在处理程序代码中读取文件的更多信息以及更多示例，请参阅 使用 Python UDF 处理程序读取文件。

此示例演示如何创建和调用使用 SnowflakeFile 类读取文件的所有者权限存储过程。

使用内联处理程序程序创建存储过程，通过传递 rb 作为 mode 实参来将输入模式指定为二进制：

CREATE OR REPLACE PROCEDURE calc_phash(file_path string)
RETURNS STRING
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
PACKAGES = ('snowflake-snowpark-python','imagehash','pillow')
HANDLER = 'run'
AS
$$
from PIL import Image
import imagehash
from snowflake.snowpark.files import SnowflakeFile

def run(ignored_session, file_path):
    with SnowflakeFile.open(file_path, 'rb') as f:
        return imagehash.average_hash(Image.open(f))
$$;

调用存储过程：

CALL calc_phash(build_scoped_file_url(@my_files, 'my_image.jpg'));

使用 Anaconda 的第三方包¶

您可以在创建 Python 存储过程时指定要安装的 Anaconda 包。要查看 Anaconda 的第三方包列表，请参阅 ` Anaconda Snowflake 通道 <https://repo.anaconda.com/pkgs/snowflake (https://repo.anaconda.com/pkgs/snowflake)>`__。这些第三方包由 Anaconda 构建和提供。您可以根据 Anaconda 服务条款的补充嵌入式软件条款免费使用 Snowflake conda 通道进行本地测试和开发。

有关限制，请参阅 限制。

开始使用¶

在开始使用 Snowflake 中的 Anaconda 提供的包之前，您必须确认 外部产品条款。

1. 登录 Snowsight。

2. 选择 Admin » Billing & Terms。

3. 在 Anaconda 部分中，选择 Enable。

4. 在 Anaconda Packages 对话框中，单击链接查看 外部产品条款页面。

5. 如果您同意条款，请选择 Acknowledge & Continue。

如果在尝试接受服务条款时出现错误，您的用户简介中可能缺少名、姓或电子邮件地址。如果您有管理员角色，请参阅 向您的用户简介添加用户详细信息，以使用 Snowsight 更新您的简介。否则，请联系管理员 更新您的账户。

显示和使用包¶

您可以通过查询 Information Schema 中的 PACKAGES 视图来显示所有可用的包及其版本信息：

select * from information_schema.packages where language = 'python';

有关更多信息，请参阅 Snowflake Python UDF 文档中 :doc:` 使用第三方包 </developer-guide/udf/python/udf-python-packages>`。

创建 Python 存储过程以自动执行 Python 工作表代码¶

从 Python 工作表创建 Python 存储过程来自动执行代码。有关编写 Python 工作表的详细信息，请参阅 在 Python 工作表中编写 Snowpark 代码。

示例¶

本节中的示例演示如何通过筛选列与字符串匹配的行的过程返回表格值。

CREATE OR REPLACE TABLE employees(id NUMBER, name VARCHAR, role VARCHAR);
INSERT INTO employees (id, name, role) VALUES (1, 'Alice', 'op'), (2, 'Bob', 'dev'), (3, 'Cindy', 'dev');

CREATE OR REPLACE PROCEDURE filterByRole(tableName VARCHAR, role VARCHAR)
RETURNS TABLE(id NUMBER, name VARCHAR, role VARCHAR)
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
PACKAGES = ('snowflake-snowpark-python')
HANDLER = 'filter_by_role'
AS
$$
from snowflake.snowpark.functions import col

def filter_by_role(session, table_name, role):
   df = session.table(table_name)
   return df.filter(col("role") == role)
$$;

CREATE OR REPLACE PROCEDURE filterByRole(tableName VARCHAR, role VARCHAR)
RETURNS TABLE()
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
PACKAGES = ('snowflake-snowpark-python')
HANDLER = 'filter_by_role'
AS
$$
from snowflake.snowpark.functions import col

def filter_by_role(session, table_name, role):
  df = session.table(table_name)
  return df.filter(col("role") == role)
$$;

CALL filterByRole('employees', 'dev');

+----+-------+------+
| ID | NAME  | ROLE |
+----+-------+------+
| 2  | Bob   | dev  |
| 3  | Cindy | dev  |
+----+-------+------+

使用工作进程运行并发任务¶

您可以使用 Python 工作进程运行并发任务。当您需要运行的并行任务利用仓库节点上的多个 CPU 核心时，您可能会发现此操作很有用。

要解决 ` Python 全局解释器锁 <https://wiki.python.org/moin/GlobalInterpreterLock (https://wiki.python.org/moin/GlobalInterpreterLock)>`_ 阻止多任务处理方法扩展到所有 CPU 核心，您可以使用单独的工作进程而不是线程来执行并发任务。

您可以使用 joblib 库的 Parallel 类对 Snowflake 仓库执行此操作，如以下示例所示。

CREATE OR REPLACE PROCEDURE joblib_multiprocessing_proc(i INT)
  RETURNS STRING
  LANGUAGE PYTHON
  RUNTIME_VERSION = 3.8
  HANDLER = 'joblib_multiprocessing'
  PACKAGES = ('snowflake-snowpark-python', 'joblib')
AS $$
import joblib
from math import sqrt

def joblib_multiprocessing(session, i):
  result = joblib.Parallel(n_jobs=-1)(joblib.delayed(sqrt)(i ** 2) for i in range(10))
  return str(result)
$$;

import joblib
joblib.parallel_backend('loky')

使用 Snowpark APIs 进行异步处理¶

以下示例说明了如何使用 Snowpark APIs 启动异步子作业，以及这些作业在不同条件下的行为方式。

CREATE OR REPLACE PROCEDURE checkStatus()
RETURNS VARCHAR
LANGUAGE PYTHON
RUNTIME_VERSION = 3.8
PACKAGES = ('snowflake-snowpark-python')
HANDLER='async_handler'
EXECUTE AS CALLER
AS $$
def async_handler(session):
    async_job = session.sql("select system$wait(60)").collect_nowait()
    return async_job.is_done()
$$;

CALL checkStatus();

+------------+
| checkStatus |
|------------|
| False      |
+------------+

CREATE OR REPLACE TABLE test_tb(c1 STRING);

CREATE OR REPLACE PROCEDURE cancelJob()
RETURNS VARCHAR
LANGUAGE PYTHON
RUNTIME_VERSION = 3.8
PACKAGES = ('snowflake-snowpark-python')
HANDLER = 'async_handler'
EXECUTE AS OWNER
AS $$
def async_handler(session):
    async_job = session.sql("insert into test_tb (select system$wait(10))").collect_nowait()
    return async_job.cancel()
$$;

CALL cancelJob();

SELECT * FROM test_tb;

+----+
| C1 |
|----|
+----+

CREATE OR REPLACE PROCEDURE blockUntilDone()
RETURNS VARCHAR
LANGUAGE PYTHON
RUNTIME_VERSION = 3.8
PACKAGES = ('snowflake-snowpark-python')
HANDLER='async_handler'
EXECUTE AS CALLER
AS $$
def async_handler(session):
    async_job = session.sql("select system$wait(5)").collect_nowait()
    return async_job.result()
$$;

CALL blockUntilDone();

+------------------------------------------+
| blockUntilDone                               |
|------------------------------------------|
| [Row(SYSTEM$WAIT(5)='waited 5 seconds')] |
+------------------------------------------+

CREATE OR REPLACE PROCEDURE earlyReturn()
RETURNS VARCHAR
LANGUAGE PYTHON
RUNTIME_VERSION = 3.8
PACKAGES = ('snowflake-snowpark-python')
HANDLER='async_handler'
EXECUTE AS CALLER
AS $$
def async_handler(session):
    async_job = session.sql("select system$wait(60)").collect_nowait()
    df = async_job.to_df()
    try:
        df.collect()
    except Exception as ex:
        return 'Error: (02000): Result for query <UUID> has expired'
$$;

CALL earlyReturn();

+------------------------------------------------------------+
| earlyReturn                                                 |
|------------------------------------------------------------|
| Error: (02000): Result for query <UUID> has expired        |
+------------------------------------------------------------+

CREATE OR REPLACE PROCEDURE earlyCancelJob()
RETURNS VARCHAR
LANGUAGE PYTHON
RUNTIME_VERSION = 3.8
PACKAGES = ('snowflake-snowpark-python')
HANDLER='async_handler'
EXECUTE AS OWNER
AS $$
def async_handler(session):
    async_job = session.sql("insert into test_tb (select system$wait(10))").collect_nowait()
$$;

CALL earlyCancelJob();
SELECT * FROM test_tb;

+----+
| C1 |
|----|
+----+