Python을 사용하여 Snowflake 동적 테이블 관리하기¶

전제 조건¶

이 항목의 예제에서는 Snowflake와 연결하고 Snowflake Python APIs 을 사용할 수 있는 Root 오브젝트를 생성하는 코드를 추가했다고 가정합니다.

예를 들어, 다음 코드는 구성 파일에 정의된 연결 매개 변수를 사용하여 Snowflake에 대한 연결을 생성합니다.

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

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

해당 코드에서는 결과 Session 오브젝트를 사용하여 API의 유형과 메서드를 사용하기 위해 Root 오브젝트를 생성합니다. 자세한 내용은 Snowflake Python APIs 을 사용하여 Snowflake에 연결 섹션을 참조하십시오.

동적 테이블 만들기¶

동적 테이블을 생성하려면 먼저 DynamicTable 오브젝트를 생성한 다음 API Root 오브젝트에서 DynamicTableCollection 오브젝트를 생성합니다. DynamicTableCollection.create 를 사용하여 Snowflake에 새 동적 테이블을 추가합니다.

다음 예제의 코드는 최소한의 필수 옵션을 지정하여 my_db 데이터베이스 및 my_schema 스키마에서 my_dynamic_table 이라는 동적 테이블을 나타내는 DynamicTable 오브젝트를 생성합니다.

from snowflake.core.dynamic_table import DynamicTable, DownstreamLag

my_dt = DynamicTable(
  name='my_dynamic_table',
  target_lag=DownstreamLag(),
  warehouse='my_wh',
  query='SELECT * FROM t',
)
dynamic_tables = root.databases['my_db'].schemas['my_schema'].dynamic_tables
dynamic_tables.create(my_dt)

이 코드는 DynamicTableCollection 변수 dynamic_tables 를 생성하고 DynamicTableCollection.create 를 사용하여 Snowflake에 새 동적 테이블을 생성합니다.

다음 예제의 코드는 현재 지원되는 모든 옵션을 지정하여 my_db 데이터베이스 및 my_schema 스키마에서 my_dynamic_table2 이라는 동적 테이블을 나타내는 DynamicTable 오브젝트를 생성합니다.

from snowflake.core.dynamic_table import DynamicTable, UserDefinedLag

root.databases['my_db'].schemas['my_schema'].dynamic_tables.create(
  DynamicTable(
      name='my_dynamic_table2',
      kind='PERMANENT',
      target_lag=UserDefinedLag(seconds=60),
      warehouse='my_wh',
      query='SELECT * FROM t',
      refresh_mode='FULL',
      initialize='ON_SCHEDULE',
      cluster_by=['id > 1'],
      comment='test table',
      data_retention_time_in_days=7,
      max_data_extension_time_in_days=7,
  )
)

from snowflake.core.dynamic_table import DynamicTableClone

root.databases['my_db'].schemas['my_schema'].dynamic_tables.create(
  DynamicTableClone(
      name='my_dynamic_table2',
      warehouse='my_wh2',
  ),
  clone_table='my_dynamic_table',
)

동적 테이블 세부 정보 얻기¶

DynamicTable 오브젝트를 반환하는 DynamicTableResource.fetch 메서드를 호출하여 동적 테이블에 대한 정보를 얻을 수 있습니다.

다음 예제의 코드는 my_db 데이터베이스와 my_schema 스키마에서 이름이 my_dynamic_table 인 동적 테이블에 대한 정보를 가져옵니다.

dynamic_table = root.databases['my_db'].schemas['my_schema'].dynamic_tables['my_dynamic_table']
dt_details = dynamic_table.fetch()
print(dt_details.to_dict())

동적 테이블 나열하기¶

DynamicTable 오브젝트의 PagedIter 반복기를 반환하는 DynamicTableCollection.iter 메서드를 사용하여 동적 테이블을 나열할 수 있습니다.

다음 예제의 코드는 my_db 데이터베이스와 my_schema 스키마에서 이름이 my 텍스트로 시작하는 동적 테이블을 나열한 다음 각 테이블의 이름을 출력합니다.

from snowflake.core.dynamic_table import DynamicTableCollection

dt_list = root.databases['my_db'].schemas['my_schema'].dynamic_tables.iter(like='my%')
for dt_obj in dt_list:
  print(dt_obj.name)

동적 테이블 이름 바꾸기¶

DynamicTableResource.swap_with 메서드를 사용하여 단일 트랜잭션에서 동적 테이블의 이름을 다른 동적 테이블과 바꿀 수 있습니다. 자세한 내용은 SWAP WITH 매개 변수 설명을 ALTER DYNAMICTABLE 에서 참조하십시오.

다음 예제의 코드는 동일한 데이터베이스와 스키마에서 my_dynamic_table 을 other_dynamic_table 로 바꿉니다.

my_table_res = root.databases['my_db'].schemas['my_schema'].tables['my_dynamic_table']
my_table_res.swap_with('other_dynamic_table')

동적 테이블 작업 수행¶

동적 테이블 새로 고침, 일시 중단 및 재개와 같은 일반적인 동적 테이블 작업을 DynamicTableResource 오브젝트를 사용하여 수행할 수 있습니다.

이러한 동적 테이블 작업에 대한 자세한 내용은 SQL 명령 참조에서 테이블, 뷰, 시퀀스 및 사용자 정의 유형 명령 섹션을 참조하십시오.

동적 테이블 리소스로 할 수 있는 일부 작업을 보여주기 위해 다음 예제의 코드는 다음을 수행합니다.

1. my_db 데이터베이스와 my_schema 스키마에서 my_dynamic_table 동적 테이블 리소스 오브젝트를 가져옵니다.

2. 동적 테이블을 새로 고칩니다.

3. 동적 테이블을 일시 중단합니다.

4. 동적 테이블을 재개합니다.

5. 동적 테이블을 삭제합니다.

6. 제거한 동적 테이블의 가장 최근 버전을 복원합니다.

my_dynamic_table_res = root.databases["my_db"].schemas["my_schema"].dynamic_tables["my_dynamic_table"]

my_dynamic_table_res.refresh()
my_dynamic_table_res.suspend()
my_dynamic_table_res.resume()
my_dynamic_table_res.drop()
my_dynamic_table_res.undrop()