Gerenciamento de tabelas dinâmicas Snowflake com Python¶

É possível usar Python para gerenciar tabelas dinâmicas Snowflake, que são um novo tipo de tabela para pipelines de processamento contínuo. Tabelas dinâmicas materializam os resultados de uma consulta especificada. Para uma visão geral deste recurso, consulte Tabelas dinâmicas.

O Snowflake Python APIs representa tabelas dinâmicas com dois tipos separados:

DynamicTable: Expõe as propriedades de uma tabela dinâmica, como nome, atraso de destino, warehouse e instrução de consulta.
DynamicTableResource: Expõe métodos que você pode usar para buscar um objeto DynamicTable correspondente, suspender e retomar a tabela dinâmica, bem como descartar a tabela dinâmica.

Pré-requisitos¶

Os exemplos neste tópico pressupõem que você tenha adicionado código para se conectar ao Snowflake e criar um objeto Root a partir do qual usar o Snowflake Python APIs.

Por exemplo, o seguinte código usa parâmetros de conexão definidos em um arquivo de configuração para criar uma conexão com o Snowflake:

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

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

Copy

Usando o objeto Session resultante, o código cria um objeto Root para usar os tipos e métodos de API. Para obter mais informações, consulte Conexão ao Snowflake com o Snowflake Python APIs.

Criação de uma tabela dinâmica¶

Para criar uma tabela dinâmica, primeiro crie um objeto DynamicTable e depois crie um objeto DynamicTableCollection a partir do objeto Root da API. Usando DynamicTableCollection.create, adicione a nova tabela dinâmica ao Snowflake.

O código no exemplo a seguir cria um objeto DynamicTable que representa uma tabela dinâmica nomeada my_dynamic_table no banco de dados my_db e no esquema my_schema, com as opções mínimas necessárias especificadas:

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)

Copy

O código cria uma variável DynamicTableCollection chamada dynamic_tables e usa DynamicTableCollection.create para criar uma nova tabela dinâmica no Snowflake.

O código no exemplo a seguir cria um objeto DynamicTable que representa uma tabela dinâmica nomeada my_dynamic_table2 no banco de dados my_db e no esquema my_schema com todas as opções atualmente possíveis especificadas:

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,
  )
)

Copy

Clonagem de uma tabela dinâmica¶

O código no exemplo a seguir cria uma nova tabela dinâmica nomeada my_dynamic_table2 com as mesmas definições de coluna e todos os dados existentes da tabela dinâmica de origem my_dynamic_table no banco de dados my_db e no esquema my_schema:

Nota

Esta operação de clonagem usa o objeto DynamicTableClone, que inclui os parâmetros opcionais target_lag e warehouse, e atualmente não oferece suporte a outros parâmetros.

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',
)

Copy

Para obter mais informações sobre esta funcionalidade, consulte CREATE DYNAMIC TABLE … CLONE.

Obtenção de detalhes dinâmicos da tabela¶

É possível obter informações sobre uma tabela dinâmica chamando o método DynamicTableResource.fetch, que retorna um objeto DynamicTable.

O código no exemplo a seguir obtém informações sobre uma tabela dinâmica nomeada my_dynamic_table no banco de dados my_db e no esquema my_schema:

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

Copy

Listagem de tabelas dinâmicas¶

É possível listar tabelas dinâmicas usando o método DynamicTableCollection.iter, que retorna um iterador PagedIter de objetos DynamicTable.

O código no exemplo a seguir lista tabelas dinâmicas cujos nomes começam com o texto my no banco de dados my_db e no esquema my_schema e, em seguida, imprime o nome de cada uma:

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)

Copy

Execução de operações de tabela dinâmica¶

É possível executar operações comuns de tabela dinâmica – como atualizar, suspender e retomar uma tabela dinâmica – com um objeto DynamicTableResource.

Para demonstrar algumas operações que você pode realizar com um recurso de tabela dinâmica, o código no exemplo a seguir faz o seguinte:

Busca o objeto de recurso da tabela dinâmica my_dynamic_table.
Atualiza a tabela dinâmica.
Suspende a tabela dinâmica.
Retoma a tabela dinâmica.
Descarta a tabela dinâmica.

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()

Copy