Como trabalhar com Snowflake no módulo de inteligência

Como puxar dados do snowflake via script

Para puxar dados do snowflake via script, seja em um notebook ou em um data app, não podemos simplesmente executar uma query no snowflake.

A abordagem a ser utilizada implica em usar a biblioteca dadosfera, a qual já vem instalada por padrão no módulo de inteligência.

Primeiramente utilize o seguinte bloco de código para importar as dependências e definir a função que puxa os dados:

from dadosfera.services.snowflake import get_snowpark_session
from typing import List

def get_snowflake_table(secret_id: str, table_identifier: str) -> List[str]:
    session = get_snowpark_session(secret_id)
    df = session.table(table_identifier)
    return df

Em seguida, defina seguinte variáveis:

• customer_name: nome do ambiente dadosfera
• table_identifier: nome do schema com um ponto separando do nome da tabela que deseja, localizada no schema referenciado

Agora utilize a função anterior para criar uma tabela snowpark:

customer_name = "DADOSFERA_ENVIROMENT"
secret_id = f"prd/root/snowflake_credentials/{customer_name}"
table_identifier = "SCHEMA.TABLE_NAME"

df = get_snowflake_table(secret_id=secret_id, table_identifier=table_identifier)

E não se esqueça de rodar um collect() para pegar os dados da tabela:

df.collect()

Como puxar os dados via step

Primeiramente, não é possível exatamente puxar dados via step. No entanto podemos utilizá-los para manipular dados dentro do Snowflake.

Para isso vá até a sua pipeline dentro do módulo de inteligência e clique em + NEW STEP:

Uma vez criado o step, selecione /step em Location e selecione o step a ser utilizado em File path. É importante ressaltar que sua primeira step deve ser o get_table_from_snowflake:

Em seguida selecione FORM como parâmetros e preencha com as informações necessárias:

Assim como na ingestão via script, passe o secret_id e o table identifier.

É importante setar o Output Type como to_outgoing_variable para poder criar um pipeline, já que o step seguinte deve ingerir dados do step anterior.

Agora crie outro step que irá alterar esses dados, por exemplo, use o snowflake_rename_columns:

Preencha com as configurações necessárias e conecte os dois steps: