Exemplos de manipuladores de UDF em Python¶

Pré-requisitos¶

Antes que seu código do manipulador Python possa ler um arquivo em um estágio, você deve fazer o seguinte para disponibilizar o arquivo para o código:

1. Crie um estágio que esteja disponível para seu manipulador.

   Você pode usar um estágio externo ou interno. Se você usar um estágio interno, isso pode ser um estágio de usuário quando planeja criar um procedimento armazenado de direitos do chamador. Caso contrário, você deve usar um estágio nomeado. No momento, o Snowflake não oferece suporte ao uso de um estágio de tabela para dependências da UDF.

   Para saber mais sobre a criação de um estágio, consulte CREATE STAGE. Para saber mais sobre a escolha de um tipo de estágio interno, consulte Escolha de um estágio interno para os arquivos locais.

   Privilégios adequados no estágio devem ser atribuídos à seguinte função, dependendo do seu caso de uso:

   Caso de uso	Função
   UDF ou procedimento armazenado de direitos do proprietário	A função que possui a UDF ou procedimento armazenado em execução.
   Procedimento armazenado de direitos do chamador	A função do usuário.

   Para obter mais informações, consulte Granting Privileges for User-Defined Functions.

2. Copie o arquivo que seu código lerá para o estágio.

   Você pode copiar o arquivo de uma unidade local para um estágio interno usando o comando PUT. Para obter mais informações sobre a preparação de arquivos com PUT, consulte Preparação de arquivos de dados de um sistema de arquivo local.

   Você pode copiar o arquivo de uma unidade local para um local de estágio externo usando qualquer uma das ferramentas fornecidas pelo seu serviço de armazenamento em nuvem. Para obter ajuda, consulte a documentação do seu serviço de armazenamento em nuvem.

Como calcular o hash perceptivo de uma imagem com um manipulador Python inline¶

Este exemplo usa SnowflakeFile para ler um par de arquivos de imagem preparados e usar o hash perceptivo (pHash) de cada arquivo para determinar a semelhança entre as imagens.

Crie uma UDF que retorna o valor de fase de uma imagem, especificando o modo de entrada como binário passando rb para o argumento mode:

CREATE OR REPLACE FUNCTION calc_phash(file_path string)
RETURNS STRING
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
PACKAGES = ('snowflake-snowpark-python','imagehash','pillow')
HANDLER = 'run'
AS
$$
from PIL import Image
import imagehash
from snowflake.snowpark.files import SnowflakeFile

def run(file_path):
  with SnowflakeFile.open(file_path, 'rb') as f:
  return imagehash.average_hash(Image.open(f))
$$;

Crie uma segunda UDF que calcula a distância entre os valores de fase de duas imagens:

CREATE OR REPLACE FUNCTION calc_phash_distance(h1 string, h2 string)
RETURNS INT
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
PACKAGES = ('imagehash')
HANDLER = 'run'
as
$$
import imagehash

def run(h1, h2):
  return imagehash.hex_to_hash(h1) - imagehash.hex_to_hash(h2)
$$;

Prepare os arquivos de imagem e atualize a tabela de diretórios:

PUT file:///tmp/image1.jpg @images AUTO_COMPRESS=FALSE;
PUT file:///tmp/image2.jpg @images AUTO_COMPRESS=FALSE;

ALTER STAGE images REFRESH;

Chame a UDFs:

SELECT
  calc_phash_distance(
    calc_phash(build_scoped_file_url(@images, 'image1.jpg')),
    calc_phash(build_scoped_file_url(@images, 'image2.jpg'))
  ) ;

Como processar um arquivo CSV com uma UDTF¶

Este exemplo usa SnowflakeFile para criar uma UDTF que extrai o conteúdo de um arquivo CSV e retorna as linhas em uma tabela.

Crie a UDTF com um manipulador inline:

CREATE FUNCTION parse_csv(file_path string)
RETURNS TABLE (col1 string, col2 string, col3 string)
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
PACKAGES = ('snowflake-snowpark-python')
HANDLER = 'csvparser'
AS
$$
from snowflake.snowpark.files import SnowflakeFile

class csvparser:
  def process(self, stagefile):
    with SnowflakeFile.open(stagefile) as f:
      for line in f.readlines():
        lineStr = line.strip()
        row = lineStr.split(",")
        try:
          # Read the columns from the line.
          yield (row[1], row[0], row[2], )
        except:
          pass
$$;

Prepare arquivo CSV e atualize a tabela de diretórios:

PUT file:///tmp/sample.csv @data_stage AUTO_COMPRESS=FALSE;

ALTER STAGE data_stage REFRESH;

Chame a UDTF, passando um URL de arquivo:

SELECT * FROM TABLE(PARSE_CSV(build_scoped_file_url(@data_stage, 'sample.csv')));

Processamento de vários arquivos¶

Você pode ler e processar vários arquivos passando a coluna RELATIVE_PATH de uma tabela de diretórios para seu manipulador. Para obter mais informações sobre a coluna RELATIVE_PATH, consulte a saída de uma consulta de tabela de diretório.

Chame uma UDF para processar vários arquivos:

O exemplo a seguir chama uma UDF dentro de uma instrução CREATE TABLE para processar cada arquivo em um estágio e, em seguida, armazenar os resultados em uma nova tabela.

Para fins de demonstração, o exemplo considera o seguinte:

• Existem vários arquivos de texto em um estágio chamado my_stage.

• Há uma UDF existente chamada get_sentiment que realiza análise de sentimento em texto não estruturado. A UDF toma um caminho para um arquivo de texto como entrada e retorna um valor que representa o sentimento.

CREATE OR REPLACE TABLE sentiment_results AS
SELECT
  relative_path
  , get_sentiment(build_scoped_file_url(@my_stage, relative_path)) AS sentiment
FROM directory(@my_stage);

Copy

Chame uma UDTF para processar vários arquivos:

Este próximo exemplo chama uma UDTF chamada parse_excel_udtf. O exemplo passa o relative_path da tabela de diretórios para o estágio chamado my_excel_stage.

SELECT t.*
FROM directory(@my_stage) d,
table(parse_excel_udtf(build_scoped_file_url(@my_excel_stage, relative_path)) t;

Copy

Como ler arquivos com URIs e URLs do estágio¶

Acesso a arquivos com SnowflakeFile exige URLs com escopo por padrão. Isso torna seu código resistente a ataques de injeção de arquivos. No entanto, você pode se referir a um local de arquivo usando um URI de estágio ou um URL de estágio em vez disso. Para isso, você deve chamar o método SnowflakeFile.open com o argumento de palavra-chave require_scoped_url = False.

Esta opção é útil quando você deseja permitir que um chamador forneça um URI que seja acessível apenas ao proprietário da UDF. Por exemplo, você pode usar um URI de estágio para acesso a arquivos se você possui uma UDF e você deseja ler em seus arquivos de configuração ou modelos de aprendizado de máquina. Não recomendamos esta opção quando você trabalha com arquivos que possuem nomes imprevisíveis, como arquivos criados com base na entrada do usuário.

Este exemplo lê um modelo de aprendizado de máquina de um arquivo e usa o modelo em uma função para executar o processamento de linguagem natural para análise de sentimento. O exemplo chama o open com require_scoped_url = False. Em ambos os formatos de localização de arquivo (URI do estágio e URL do estágio), o proprietário da UDF deve ter acesso ao arquivo de modelo.

Crie a UDF com um manipulador inline:

CREATE OR REPLACE FUNCTION extract_sentiment(input_data string)
RETURNS STRING
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
PACKAGES = ('snowflake-snowpark-python','scikit-learn')
HANDLER = 'run'
AS
$$
from snowflake.snowpark.files import SnowflakeFile
from sklearn.linear_model import SGDClassifier
import pickle

def run(input_data):
  model_file = '@models/NLP_model.pickle'
  # Specify 'mode = rb' to open the file in binary mode.
  with SnowflakeFile.open(model_file, 'rb', require_scoped_url = False) as f:
    model = pickle.load(f)
    return model.predict([input_data])[0]
$$;

Prepare o arquivo de modelo e atualize a tabela de diretórios:

PUT file:///tmp/NLP_model.pickle @models AUTO_COMPRESS=FALSE;

ALTER STAGE models REFRESH;

Como alternativa, você pode especificar a UDF com o URL do estágio do modelo para extrair o sentimento.

Por exemplo, crie uma UDF com um manipulador inline que especifica um arquivo usando um URL do estágio:

CREATE OR REPLACE FUNCTION extract_sentiment(input_data string)
RETURNS STRING
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
PACKAGES = ('snowflake-snowpark-python','scikit-learn')
HANDLER = 'run'
AS
$$
from snowflake.snowpark.files import SnowflakeFile
from sklearn.linear_model import SGDClassifier
import pickle

def run(input_data):
  model_file = 'https://my_account/api/files/my_db/my_schema/models/NLP_model.pickle'
  # Specify 'rb' to open the file in binary mode.
  with SnowflakeFile.open(model_file, 'rb', require_scoped_url = False) as f:
    model = pickle.load(f)
    return model.predict([input_data])[0]
$$;

Chame a UDF com os dados de entrada:

SELECT extract_sentiment('I am writing to express my interest in a recent posting made.');

Copy