PythonによるSnowflake仮想ウェアハウスの管理¶

Pythonを使用して、SnowflakeのコンピューティングリソースのクラスタであるSnowflake仮想ウェアハウスを管理できます。ウェアハウスの概要については、仮想ウェアハウスをご参照ください。

Snowflake Python API は、2つの別々のタイプでウェアハウスを表します。

Warehouse: ウェアハウスの名前、サイズ、タイプ、自動再開と自動中断の設定などのプロパティを公開します。
WarehouseResource: 対応する Warehouse オブジェクトのフェッチ、ウェアハウスの中断と再開、ウェアハウスの削除に使用できるメソッドを公開します。

このトピックの例では、Snowflakeと接続するコードを追加して Root オブジェクトを作成し、そこからSnowflake Python API を使用することを想定しています。詳細については、 Snowflake Python API によるSnowflakeへの接続をご参照ください。

次の例のコードは、構成ファイルで定義された接続パラメーターを使用して、Snowflakeへの接続を作成します。出来上がった Session オブジェクトを使って、コードは API のタイプとメソッドを使う Root オブジェクトを作成します。

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

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

Copy

ウェアハウスを作成する¶

ウェアハウスを作成するには、まず Warehouse オブジェクトを作成します。次に、 API Root オブジェクトから WarehouseCollection オブジェクトを作成します。 WarehouseCollection.create を使用して、新しいウェアハウスをSnowflakeに追加します。

次の例のコードは、 my_wh というウェアハウスを表す Warehouse オブジェクトを作成します。 WarehouseCollection 変数 warehouses を作成し、 WarehouseCollection.create を使用してSnowflakeに新しいウェアハウスを作成します。

from snowflake.core import Root
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

ウェアハウスの詳細を取得する¶

Warehouse オブジェクトを返す WarehouseResource.fetch メソッドを呼び出すことで、ウェアハウスに関する情報を取得できます。

次の例のコードは、 my_wh というウェアハウスに関する情報を取得します。

from snowflake.core import Root
from snowflake.core.warehouse import Warehouse

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

Copy

ウェアハウスを作成または更新する¶

ウェアハウスを表す Warehouse オブジェクトのプロパティを設定し、 WarehouseResource.create_or_update メソッドでオブジェクトを Snowflakeに渡すことで、既存のウェアハウスの特性を更新できます。

ウェアハウスを作成するときに、新しいウェアハウスを記述する Warehouse オブジェクトを渡すこともできます。

次の例のコードは、ウェアハウスの名前、サイズ、自動中断設定を設定した上で、Snowflakeでウェアハウスを更新し、まだ存在しない場合はウェアハウスを作成します。このケースでは、 my_wh ウェアハウスのサイズを LARGE に、自動中断の設定を 1800 に変更します。

from snowflake.core import Root
from snowflake.core.warehouse import Warehouse

my_wh = Warehouse(
  name="my_wh",
  warehouse_size="LARGE",
  auto_suspend=1800,
)
my_wh_res = root.warehouses["my_wh"]
my_wh_res.create_or_update(my_wh)

Copy

ウェアハウスを一覧表示する¶

WarehouseCollection.iter メソッド使用してウェアハウスを一覧表示できます。このメソッドは Warehouse オブジェクトの PagedIter 反復子を返します。

次の例のコードは、名前に「my」というテキストを含むウェアハウスを一覧表示し、それぞれの名前を印刷します。

from snowflake.core import Root
from snowflake.core.warehouse import Warehouse, 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 ウェアハウスを中断および再開し、ウェアハウス上で実行中またはキューに入っているすべてのクエリを中止し、ウェアハウスを削除します。

from snowflake.core import Root
from snowflake.core.warehouse import Warehouse

my_wh_res = root.warehouses["my_wh"]

my_wh_res.suspend()
my_wh_res.resume()
my_wh_res.abort_all_queries()
my_wh_res.delete()

Copy