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

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

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

  • Warehouse: ウェアハウスの名前、サイズ、タイプ、自動再開と自動中断の設定などのプロパティを公開します。

  • WarehouseResource: 対応する Warehouse オブジェクトの取得、ウェアハウスの中断と再開、ウェアハウスの削除に使用できるメソッドを公開します。

前提条件

このトピックの例では、Snowflakeと接続するコードを追加して Root オブジェクトを作成し、そこからSnowflake Python 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 オブジェクトを使って、コードは API のタイプとメソッドを使う Root オブジェクトを作成します。詳細については、 Snowflake Python APIs によるSnowflakeへの接続 をご参照ください。

ウェアハウスの作成

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

次の例のコードは、 my_wh というウェアハウスを表す Warehouse オブジェクトを作成します。

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に新しいウェアハウスを作成します。

ウェアハウスの詳細の取得

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

次の例のコードは、 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 メソッドは、明示的に定義していない Warehouse プロパティにはデフォルト値を使用します。例えば、 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 に変更します。

ウェアハウスのリスト

Warehouse オブジェクトの PagedIter 反復子を返す WarehouseCollection.iter メソッドを使用して、ウェアハウスをリストすることができます。

次の例のコードは、名前に 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
言語: 日本語