用 Python 管理 Snowflake 虚拟仓库

您可以使用 Python 管理 Snowflake 虚拟仓库，虚拟仓库是 Snowflake 中的计算资源集群。有关仓库的概述，请参阅 虚拟仓库。

Snowflake Python APIs 使用两种不同类型来表示仓库：

  • Warehouse：显示仓库的属性，如名称、大小、类型、自动恢复和自动暂停设置。

  • WarehouseResource：显示可用于提取相应的 Warehouse 对象、暂停和恢复仓库以及删除仓库的方法。

先决条件

在本主题中的示例中，假设您已添加了用来连接 Snowflake 和创建 Root 对象以使用 Snowflake Python APIs 的代码。

例如，以下代码使用配置文件中定义的连接参数来创建与 Snowflake 的连接：

from snowflake.core import Root
from snowflake.snowpark import Session

session = Session.builder.config("connection_name", "myconnection").create()
root = Root(session)
Copy

代码可通过生成的 Session 对象创建 Root 对象，从而使用 API 的类型和方法。有关更多信息，请参阅 使用 Snowflake Python APIs 连接到 Snowflake。

创建仓库

要创建仓库，请先创建 Warehouse 对象，然后根据 API Root 对象创建 WarehouseCollection 对象。使用 WarehouseCollection.create 将新仓库添加到 Snowflake。

以下示例中的代码会创建一个 Warehouse 对象，其表示名为 my_wh 的仓库：

from snowflake.core.warehouse import Warehouse

my_wh = Warehouse(
  name="my_wh",
  warehouse_size="SMALL",
  auto_suspend=600,
)
warehouses = root.warehouses
warehouses.create(my_wh)
Copy

该代码会创建一个 WarehouseCollection 变量 warehouses ，并使用 WarehouseCollection.create 在 Snowflake 中创建了一个新仓库。

获取仓库详细信息

您可以通过调用 WarehouseResource.fetch 方法获取有关的仓库信息，该方法会返回一个 Warehouse 对象。

以下示例中的代码会获取有关名为 my_wh 的仓库的信息：

my_wh = root.warehouses["my_wh"].fetch()
print(my_wh.to_dict())
Copy

创建或更改仓库

您可以设置 Warehouse 对象的属性并将其传递给 WarehouseResource.create_or_alter 方法以创建仓库（如果仓库不存在）或者根据仓库定义更改仓库（如果仓库存在）。create_or_alter 的行为旨在实现幂等性，这意味着无论在调用该方法之前仓库是否存在，生成的仓库对象都是一样的。

备注

对于任何没有显式定义的 仓库 属性，create_or_alter 方法会使用默认值。例如，如果您没有设置 auto_suspend，即使仓库以前存在不同的值，其值也会默认为 None。

以下示例中的代码会更新 my_wh 仓库的大小和自动暂停设置，然后在 Snowflake 上更改仓库。

from snowflake.core.warehouse import Warehouse

my_wh = root.warehouses["my_wh"].fetch()
my_wh.warehouse_size = "LARGE"
my_wh.auto_suspend = 1800

my_wh_res = root.warehouses["my_wh"]
my_wh_res.create_or_alter(my_wh)
Copy

在此示例中，它将 my_wh 仓库的大小更改为 LARGE，并将其自动暂停设置更改为 ``1800``（如果您之前使用不同的属性创建它）。

列出仓库

您可以使用 WarehouseCollection.iter 方法列出仓库，该方法会返回 Warehouse 对象的 PagedIter 迭代器。

以下示例中的代码会列出名称中包含 my 文本的仓库，并打印每个仓库的名称：

from snowflake.core.warehouse import WarehouseCollection

warehouses: WarehouseCollection = root.warehouses
wh_iter = warehouses.iter(like="my%")  # returns a PagedIter[Warehouse]
for wh_obj in wh_iter:
  print(wh_obj.name)
Copy

执行仓库操作

您可以通过 WarehouseResource 对象执行常见的仓库操作，如暂停和恢复仓库以及中止仓库上的所有查询。

以下示例中的代码会暂停和恢复 my_wh 仓库，中止仓库上所有正在运行或排队的查询，然后删除仓库：

my_wh_res = root.warehouses["my_wh"]

my_wh_res.suspend()
my_wh_res.resume()
my_wh_res.abort_all_queries()
my_wh_res.drop()
Copy
语言: 中文