PythonによるSnowflakeデータベース、スキーマ、テーブル、ビューの管理

Pythonを使ってSnowflakeデータベース、スキーマ、テーブル、ビューを管理することができます。Snowflakeでのデータの管理と操作の詳細については、 データベース、テーブル、およびビュー - 概要 をご参照ください。

前提条件

このトピックの例では、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への接続 をご参照ください。

データベースの管理

Snowflakeでデータベースを管理できます。 Snowflake Python APIs は、データベースを2つの別々のタイプで表します。

  • Database: データベースの名前などのプロパティを公開します。

  • DatabaseResource: 対応する Database オブジェクトを取得し、データベースをドロップするために使用できるメソッドを公開します。

トピックス

データベースの作成

DatabaseCollection.create メソッドを呼び出し、作成するデータベースを表す Database オブジェクトを渡すと、データベースを作成できます。データベースを作成するには、まずデータベース名を指定する Database オブジェクトを作成します。

次の例のコードは、 my_db という名前のデータベースを表す Database オブジェクトを作成してから、 Database オブジェクトを DatabaseCollection.create メソッドに渡してデータベースを作成します。

from snowflake.core.database import Database

my_db = Database(name="my_db")
root.databases.create(my_db)
Copy

データベースの詳細の取得

Database オブジェクトを返す DatabaseResource.fetch メソッドを呼び出すことで、データベースに関する情報を取得できます。

次の例のコードは、 my_db というデータベースに関する情報を取得します。

my_db = root.databases["my_db"].fetch()
print(my_db.to_dict())
Copy

データベースのリスト

PagedIter 反復子を返す iter メソッドを使用して、データベースをリストすることができます。

次の例のコードは、名前が my で始まるデータベースを一覧表示します。

databases = root.databases.iter(like="my%")
for database in databases:
  print(database.name)
Copy

データベースのドロップ

DatabaseResource.drop メソッドを使用してデータベースをドロップできます。

以下の例のコードは、 my_db のデータベースをドロップします。

my_db_res = root.databases["my_db"]
my_db_res.drop()
Copy

スキーマの管理

Snowflakeでスキーマを管理できます。スキーマはデータベースレベルのオブジェクトです。スキーマを作成または参照するときは、そのデータベースのコンテキスト内で行います。

Snowflake Python APIs は、スキーマを2つの別々のタイプで表します。

  • Schema: スキーマの名前などのプロパティを公開します。

  • SchemaResource: 対応する Schema オブジェクトを取得し、スキーマをドロップするために使用できるメソッドを公開します。

トピックス

スキーマの作成

スキーマを作成するには、まずスキーマ名を指定する Schema オブジェクトを作成します。

次の例のコードは、 my_schema という名前のスキーマを表す Schema オブジェクトを作成します。

from snowflake.core.schema import Schema

my_schema = Schema(name="my_schema")
root.databases["my_db"].schemas.create(my_schema)
Copy

次に、コードは Schema オブジェクトを SchemaCollection.create メソッドに渡して、 my_db データベースにスキーマを作成します。

スキーマの詳細を取得する

Schema オブジェクトを返す SchemaResource.fetch メソッドを呼び出すことで、スキーマの情報を取得できます。

次の例のコードは、 my_schema スキーマを表す Schema オブジェクトを取得します。

my_schema = root.databases["my_db"].schemas["my_schema"].fetch()
print(my_schema.to_dict())
Copy

スキーマのリスト

iter メソッドを使用して指定したデータベース内のスキーマを一覧表示できます。このメソッドは Schema オブジェクトの PagedIter 反復子を返します。

次の例のコードは、 my_db データベースのスキーマ名をリストします。

schema_list = root.databases["my_db"].schemas.iter()
for schema_obj in schema_list:
  print(schema_obj.name)
Copy

スキーマのドロップ

SchemaResource.drop メソッドを使用してスキーマをドロップできます。

以下の例のコードは、 my_schema のスキーマをドロップします。

my_schema_res = root.databases["my_db"].schemas["my_schema"]
my_schema_res.drop()
Copy

標準テーブルの管理

Snowflakeで標準テーブルを管理できます。テーブルはスキーマレベルのオブジェクトです。テーブルを作成または参照するときは、そのデータベースのコンテキスト内で行います。

Snowflake Python APIs は、テーブルを2つの別々のタイプで表します。

  • Table: テーブルの名前や列などのプロパティを公開します。

  • TableResource: 対応する Table オブジェクトを取得し、テーブルのプロパティの更新、テーブルのドロップに使用できるメソッドを公開します。

トピックス

テーブルの作成

テーブルを作成するには、まずテーブル名、列名、列のデータ型を指定する Table オブジェクトを作成します。

次の例のコードは、指定された列を持つ my_table という名前のテーブルを表す Table オブジェクトを作成します。

from snowflake.core.table import Table, TableColumn

my_table = Table(
  name="my_table",
  columns=[TableColumn(name="c1", datatype="int", nullable=False),
           TableColumn(name="c2", datatype="string")]
)
root.databases["my_db"].schemas["my_schema"].tables.create(my_table)
Copy

次に、コードは Table オブジェクトを TableCollection.create メソッドに渡して、 my_db データベースと my_schema スキーマにテーブルを作成します。

テーブルの詳細の取得

Table オブジェクトを返す TableResource.fetch メソッドを呼び出すことで、テーブルに関する情報を取得できます。

次の例のコードは、 my_table という名前のテーブルの情報を取得します。

my_table = root.databases["my_db"].schemas["my_schema"].tables["my_table"].fetch()
print(my_table.to_dict())
Copy

テーブルの作成と変更

Table オブジェクトのプロパティを設定し、 TableResource.create_or_alter メソッドに渡すことで、テーブルが存在しない場合は作成し、存在する場合はテーブル定義に従って変更することができます。 create_or_alter の動作は、べき等であることを目的としています。つまり、メソッドを実行する前にテーブルが存在していたかどうかに関係なく、結果として得られるテーブル・オブジェクトは同じものになります。

注釈

create_or_alter メソッドは、明示的に定義していない テーブル プロパティにデフォルト値を使用します。例えば、 data_retention_time_in_days を設定しない場合、以前は異なる値でテーブルが存在していたとしても、その値はデフォルトで None になります。

次の例のコードは、 my_table テーブルにデータ型 int の c3 という新しい列を追加してから、Snowflakeのテーブルを更新します。

from snowflake.core.table import PrimaryKey, TableColumn

my_table = root.databases["my_db"].schemas["my_schema"].tables["my_table"].fetch()
my_table.columns.append(TableColumn(name="c3", datatype="int", nullable=False, constraints=[PrimaryKey()]))

my_table_res = root.databases["my_db"].schemas["my_schema"].tables["my_table"]
my_table_res.create_or_alter(my_table)
Copy

テーブルのリスト

iter メソッドを使用して、指定したスキーマにあるテーブルをリストできます。これは、 Table オブジェクトの PagedIter 反復子を返します。

次の例のコードは、名前が my で始まるテーブルを一覧表示します。

tables = root.databases["my_db"].schemas["my_schema"].tables.iter(like="my%")
for table_obj in tables:
  print(table_obj.name)
Copy

テーブルのドロップ

TableResource.drop メソッドを使用してテーブルをドロップできます。

以下の例のコードは、 my_table のテーブルをドロップします。

my_table_res = root.databases["my_db"].schemas["my_schema"].tables["my_table"]
my_table_res.drop()
Copy

イベントテーブルの管理

Snowflakeイベントテーブルは、Snowflakeがテレメトリーデータを収集できる、事前定義された一連の列を持つ特別な種類のデータベーステーブルです。詳細については、 イベントテーブルの概要 をご参照ください。

Snowflake Python APIs は、イベント・テーブルを2つの別々のタイプで表します。

  • EventTable: イベントテーブルの名前、データ保持時間、最大データ延長時間、変更追跡オプションなどのプロパティを公開します。

  • EventTableResource: 対応する EventTable オブジェクトの取得、イベントテーブルの名前の変更、イベントテーブルのドロップに使用できるメソッドを公開します。

トピックス

イベントテーブルの作成

ウェアハウスを作成するには、まず EventTable オブジェクトを作成して、 API Root オブジェクトから EventTableCollection オブジェクトを作成します。 EventTableCollection.create を使用して、新しいイベントテーブルをSnowflakeに追加します。

以下の例のコードは、 my_event_table という名前のイベント・テーブルを表す EventTable オブジェクトを、指定されたパラメーターで作成します。

from snowflake.core.event_table import EventTable

event_table = EventTable(
  name="my_event_table",
  data_retention_time_in_days = 3,
  max_data_extension_time_in_days = 5,
  change_tracking = True,
  default_ddl_collation = 'EN-CI',
  comment = 'CREATE EVENT TABLE'
)

event_tables = root.databases["my_db"].schemas["my_schema"].event_tables
event_tables.create(my_event_table)
Copy

コードは、 EventTableCollection 変数 event_tables を作成し、 EventTableCollection.create を使用してSnowflakeに新しいウェアハウスを作成します。

イベントテーブルの詳細を取得する

EventTable オブジェクトを返す EventTableResource.fetch メソッドを呼び出すことで、テーブルに関する情報を取得できます。

次の例のコードは、 my_event_table という名前のイベント・テーブルの情報を取得します。

my_event_table = root.databases["my_db"].schemas["my_schema"].event_tables["my_event_table"].fetch()
print(my_event_table.to_dict())
Copy

イベントテーブルの一覧表示

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

次の例のコードは、 my_db データベースと my_schema スキーマの my で始まる名前のイベント・テーブルを一覧表示します。

from snowflake.core.event_table import EventTableCollection

event_tables: EventTableCollection = root.databases["my_db"].schemas["my_schema"].event_tables
event_table_iter = event_tables.iter(like="my%")  # returns a PagedIter[EventTable]
for event_table_obj in event_table_iter:
  print(event_table_obj.name)
Copy

次の例のコードも、 my で始まる名前のイベント・テーブルをリストしていますが、 like の代わりに starts_with パラメーターを使用しています。この例では、結果の数を 10 に制限するために、オプションのパラメーター show_limit=10 も設定しています。

event_tables: EventTableCollection = root.databases["my_db"].schemas["my_schema"].event_tables
event_table_iter = event_tables.iter(starts_with="my", show_limit=10)
for event_table_obj in event_table_iter:
  print(event_table_obj.name)
Copy

イベントテーブルの操作の実行

EventTableResource オブジェクトを使用して、イベント・テーブルの名前の変更やイベント・テーブルのドロップなど、一般的なイベント・テーブルの操作を行うことができます。

注釈

現在のところ、 ALTER TABLE （イベントテーブル） の RENAME 機能のみがサポートされています。

RENAME はデフォルトのイベントテーブル、 SNOWFLAKE.TELEMETRY.EVENTS ではサポートされていません。

イベント・テーブル・リソースでできる操作を示すために、次の例のコードでは次のようにしています。

  1. my_event_table イベント テーブル リソース オブジェクトを取得します。

  2. イベントテーブルの名前を変更する

  3. イベントテーブルをドロップします。

my_event_table_res = root.databases["my_db"].schemas["my_schema"].event_tables["my_event_table"]

my_event_table_res.rename("my_other_event_table")
my_event_table_res.drop()
Copy

ビューの管理

Snowflakeではビューを管理することができます。ビューはスキーマレベルオブジェクトで、ビューを使用すると、クエリの結果がテーブルであるかのようにアクセスできます。ビューを作成または参照するときは、そのスキーマのコンテキスト内で行います。

注釈

ALTER VIEW は現在サポートされていません。

Snowflake Python APIs は、ビューを2つの別々のタイプで表します。

  • View: ビューの名前、列、 SQL クエリステートメントなど、ビューのプロパティを公開します。

  • ViewResource: 対応する View オブジェクトを取得し、ビューをドロップするために使用できるメソッドを公開します。

トピックス

ビューの作成

ビューを作成するには、まずビュー名、列、 SQL クエリステートメントを指定する View オブジェクトを作成します。

次の例のコードは、指定された列と SQL クエリを持つ my_view という名前のビューを表す View オブジェクトを作成します。

from snowflake.core.view import View, ViewColumn

my_view = View(
  name="my_view",
  columns=[
      ViewColumn(name="c1"), ViewColumn(name="c2"), ViewColumn(name="c3"),
  ],
  query="SELECT * FROM my_table",
)

root.databases["my_db"].schemas["my_schema"].views.create(my_view)
Copy

次に、コードは View オブジェクトを ViewCollection.create メソッドに渡して、 my_db データベースと my_schema スキーマにビューを作成します。

ビューの詳細を取得する

View オブジェクトを返す ViewResource.fetch メソッドを呼び出すことで、ビューに関する情報を取得できます。

次の例のコードは、 my_view ビューを表す View オブジェクトを取得します。

my_view = root.databases["my_db"].schemas["my_schema"].views["my_view"].fetch()
print(my_view.to_dict())
Copy

リストの表示

iter メソッドを使用して指定したデータベース内のビューを一覧表示できます。このメソッドは View オブジェクトの PagedIter 反復子を返します。

次の例のコードは、 my_db データベースと my_schema スキーマの my で始まる名前のビューを一覧表示します。

view_list = root.databases["my_db"].schemas["my_schema"].views.iter(like="my%")
for view_obj in view_list:
  print(view_obj.name)
Copy

以下の例のコードも、名前が my で始まるビューをリストアップしていますが、 like の代わりに starts_with パラメーターを使用しています。この例では、結果の数を 10 に制限するために、オプションのパラメーター show_limit=10 も設定しています。

view_list = root.databases["my_db"].schemas["my_schema"].views.iter(starts_with="my", show_limit=10)
for view_obj in view_list:
  print(view_obj.name)
Copy

ビューのドロップ

ViewResource.drop メソッドを使用してビューをドロップすることができます。

以下の例のコードは、 my_view のビューをドロップします。

my_view_res = root.databases["my_db"].schemas["my_schema"].views["my_view"]
my_view_res.drop()
Copy
言語: 日本語