UDFs tabulares (UDTFs) de Java¶

A classe do manipulador¶

O Snowflake interage com a UDTF principalmente invocando os seguintes métodos da classe do manipulador:

• O inicializador (o construtor).

• O método por linha (process).

• O método finalizador (endPartition).

A classe do manipulador pode conter métodos adicionais necessários para apoiar esses três métodos.

A classe do manipulador também contém um método getOutputClass, que é descrito mais adiante.

Gerar uma exceção de qualquer método na classe do manipulador (ou na classe da linha de saída) faz com que o processamento seja interrompido. A consulta que chamou a UDTF falha com uma mensagem de erro.

import java.util.stream.Stream;

...

public Stream<OutputRow> process(String v) {
  return Stream.of(new OutputRow(v));
}

...

public Stream<OutputRow> process(int number) {
  if (inputNumber < 1) {
    return Stream.empty();
  }
  return Stream.of(new OutputRow(number));
}

A classe da linha de saída¶

O Snowflake usa a classe da linha de saída para ajudar a especificar as conversões entre tipos de dados de Java e de SQL.

Quando uma UDTF de Java retorna uma linha, o valor em cada coluna da linha deve ser convertido do tipo de dados de Java para o tipo de dados de SQL correspondente. Os tipos de dados de SQL são especificados na cláusula RETURNS da instrução CREATE FUNCTION. Entretanto, o mapeamento entre os tipos de dados de Java e de SQL não tem uma correspondência exata; portanto, o Snowflake precisa saber o tipo de dados de Java para cada coluna retornada. (Para obter mais informações sobre o mapeamento de tipos de dados de SQL e Java, consulte Mapeamentos de tipos de dados SQL-Java para tipos de parâmetros e retorno).

Uma UDTF de Java especifica os tipos de dados de Java das colunas de saída definindo uma classe de linha de saída. Cada linha retornada da UDTF é retornada como uma instância da classe da linha de saída. Cada instância da classe da linha de saída contém um campo público para cada coluna de saída. O Snowflake lê os valores dos campos públicos de cada instância da classe da linha de saída, converte os valores de Java para valores de SQL e cria uma linha de saída de SQL contendo esses valores.

Os valores em cada instância da classe da linha de saída são definidos chamando o construtor da classe da linha de saída. O construtor aceita parâmetros que correspondem às colunas de saída e depois define os campos públicos para esses parâmetros.

O código abaixo define uma classe de linha de saída de exemplo:

class OutputRow {

  public String name;
  public int id;

  public OutputRow(String pName, int pId) {
    this.name = pName;
    this.id = pId
  }

}

As variáveis públicas especificadas por essa classe devem corresponder às colunas especificadas na cláusula RETURNS TABLE (...) da instrução CREATE FUNCTION. Por exemplo, a classe OutputRow acima corresponde à cláusula RETURNS abaixo:

CREATE FUNCTION F(...)
    RETURNS TABLE(NAME VARCHAR, ID INTEGER)
    ...

A classe da linha de saída é utilizada da seguinte forma:

• A classe do manipulador usa a classe da linha de saída para especificar o tipo de retorno do método process e do método endPartition. A classe do manipulador também usa a classe da linha de saída para construir valores retornados. Por exemplo:

  public Stream<OutputRow> process(String v) {
    ...
    return Stream.of(new OutputRow(...));
  }
  
  public Stream<OutputRow> endPartition() {
    ...
    return Stream.of(new OutputRow(...));
  }
  

• A classe de linha de saída também é usada no método getOutputClass da classe do manipulador, que é um método estático que o Snowflake chama para aprender os tipos de dados de Java a partir das saídas:

  public static Class getOutputClass() {
    return OutputRow.class;
  }
  

A geração de uma exceção de qualquer método na classe de linha de saída (ou na classe do manipulador) faz com que o processamento pare. A consulta que chamou a UDTF falha com uma mensagem de erro.

Resumo dos requisitos¶

O código da UDTF de Java deve atender aos seguintes requisitos:

• O código deve definir uma classe de linha de saída.

• A classe do manipulador da UDTF deve incluir um método público chamado process que retorna um Stream de <output_row_class>, onde Stream é definido em java.util.stream.stream.Stream.

• A classe do manipulador da UDTF deve definir um método estático público chamado getOutputClass, que deve retornar <output_row_class>.class.

Se o de código Java não atender a esses requisitos, a criação ou execução da UDTF falha:

• Se a sessão tiver um warehouse ativo no momento em que a instrução CREATE FUNCTION for executada, o Snowflake detectará violações quando a função for criada.

• Se a sessão não tiver um warehouse ativo no momento em que a instrução CREATE FUNCTION for executada, o Snowflake detectará violações quando a função for chamada.

Particionamento explícito¶

SELECT *
    FROM stocks_table AS st,
         TABLE(my_udtf(st.symbol, st.transaction_date, st.price) OVER (PARTITION BY st.symbol))

SELECT *
    FROM stocks_table AS st,
         TABLE(my_udtf(st.symbol, st.transaction_date, st.price) OVER (PARTITION BY 1))

SELECT *
     FROM stocks_table AS st,
          TABLE(my_udtf(st.symbol, st.transaction_date, st.price) OVER (PARTITION BY st.symbol ORDER BY st.transaction_date))

SELECT *
    FROM stocks_table AS st,
         TABLE(my_udtf(st.symbol, st.transaction_date, st.price) OVER (PARTITION BY st.symbol ORDER BY st.transaction_date))
    ORDER BY st.symbol, st.transaction_date, st.transaction_time;

SELECT * FROM udtf_table, TABLE(my_func(col1) OVER (PARTITION BY udtf_table.col2 * 2));   -- NO!

Particionamento implícito¶

Se uma função de tabela não particiona explicitamente as linhas usando uma cláusula PARTITION BY, o Snowflake normalmente divide as linhas implicitamente para usar processamento paralelo para melhorar o desempenho.

O número de partições normalmente é baseado em fatores como o tamanho do warehouse que processa a função e a cardinalidade da relação de entrada. As linhas normalmente são atribuídas a partições específicas com base em fatores como a localização física das linhas (por exemplo, por micropartição), de modo que o agrupamento não tem significado.

Quando executado com particionamento implícito, o código do usuário não pode fazer suposições sobre partições. A execução com particionamento implícito é mais útil quando a UDTF só precisa olhar as linhas isoladamente para produzir sua saída e nenhum estado é agregado entre as linhas. Nesse caso, o código provavelmente não precisa de um construtor ou de um método endPartition.

Se a UDF inclui um método finalizador (por exemplo, endPartition para UDFs de Java), o finalizador é chamado em cada partição, independentemente de os dados terem sido particionados explícita ou implicitamente. Se os dados não forem particionados de forma significativa, a saída do finalizador pode não ser significativa.

Notas de uso para particionamento¶

• Para melhorar o desempenho, o Snowflake normalmente executa várias instâncias do código do manipulador da UDTF em paralelo. Cada partição de linhas é passada para uma única instância da UDTF.

  Embora cada partição seja processada por apenas uma instância de UDTF, o inverso não é necessariamente verdadeiro — uma única instância de UDTF pode processar várias partições sequencialmente. Portanto, é importante usar o inicializador e o finalizador para inicializar e limpar cada partição, de forma a evitar passar valores acumulados do processamento de uma partição para o processamento de outra partição.

Chamada sem particionamento explícito¶

Este exemplo mostra como criar uma UDTF. Este exemplo retorna duas cópias de cada entrada e retorna uma linha adicional para cada partição.

create function return_two_copies(v varchar)
returns table(output_value varchar)
language java
handler='TestFunction'
target_path='@~/TestFunction.jar'
as
$$

  import java.util.stream.Stream;

  class OutputRow {

    public String output_value;

    public OutputRow(String outputValue) {
      this.output_value = outputValue;
    }

  }


  class TestFunction {

    String myString;

    public TestFunction()  {
      myString = "Created in constructor and output from endPartition()";
    }

    public static Class getOutputClass() {
      return OutputRow.class;
    }

    public Stream<OutputRow> process(String inputValue) {
      // Return two rows with the same value.
      return Stream.of(new OutputRow(inputValue), new OutputRow(inputValue));
    }

    public Stream<OutputRow> endPartition() {
      // Returns the value we initialized in the constructor.
      return Stream.of(new OutputRow(myString));
    }

  }

$$;

Este exemplo mostra como chamar uma UDTF. Para manter este exemplo simples, a instrução passa um valor literal em vez de uma coluna e omite a cláusula OVER().

SELECT output_value
   FROM TABLE(return_two_copies('Input string'));
+-------------------------------------------------------+
| OUTPUT_VALUE                                          |
|-------------------------------------------------------|
| Input string                                          |
| Input string                                          |
| Created in constructor and output from endPartition() |
+-------------------------------------------------------+

Este exemplo chama a UDTF com valores lidos de outra tabela. Cada vez que o método process é chamado, ele recebe um valor da coluna city_name da linha atual da tabela cities_of_interest. Como acima, a cláusula UDTF é chamada sem uma cláusula OVER() explícita.

Crie uma tabela simples para ser usada como fonte de entradas:

CREATE TABLE cities_of_interest (city_name VARCHAR);
INSERT INTO cities_of_interest (city_name) VALUES
    ('Toronto'),
    ('Warsaw'),
    ('Kyoto');

Chame a UDTF de Java:

SELECT city_name, output_value
   FROM cities_of_interest,
       TABLE(return_two_copies(city_name))
   ORDER BY city_name, output_value;
+-----------+-------------------------------------------------------+
| CITY_NAME | OUTPUT_VALUE                                          |
|-----------+-------------------------------------------------------|
| Kyoto     | Kyoto                                                 |
| Kyoto     | Kyoto                                                 |
| Toronto   | Toronto                                               |
| Toronto   | Toronto                                               |
| Warsaw    | Warsaw                                                |
| Warsaw    | Warsaw                                                |
| NULL      | Created in constructor and output from endPartition() |
+-----------+-------------------------------------------------------+

A seção de exemplos na documentação para UDTFs de JavaScript contém exemplos mais complexos de consultas que chamam UDTFs com valores de tabelas.

Se a instrução não especificar explicitamente o particionamento, o mecanismo de execução do Snowflake usa particionamento implícito.

Se houver apenas uma partição, o método endPartition é chamado apenas uma vez e a saída da consulta inclui apenas uma linha que contém o valor Created in constructor and output from endPartition(). Se os dados forem agrupados em diferentes números de partições durante diferentes execuções da instrução, o método endPartition é chamado em números de vezes diferentes, e a saída contém números diferentes de cópias dessa linha.

Para obter mais informações, consulte partição implícita.

Chamada com partição explícita¶

UDTFs de Java também podem ser chamadas usando partição explícita.

SELECT city_name, output_value
   FROM cities_of_interest,
       TABLE(return_two_copies(city_name) OVER (PARTITION BY city_name))
   ORDER BY city_name, output_value;
+-----------+-------------------------------------------------------+
| CITY_NAME | OUTPUT_VALUE                                          |
|-----------+-------------------------------------------------------|
| Kyoto     | Created in constructor and output from endPartition() |
| Kyoto     | Kyoto                                                 |
| Kyoto     | Kyoto                                                 |
| Toronto   | Created in constructor and output from endPartition() |
| Toronto   | Toronto                                               |
| Toronto   | Toronto                                               |
| Warsaw    | Created in constructor and output from endPartition() |
| Warsaw    | Warsaw                                                |
| Warsaw    | Warsaw                                                |
+-----------+-------------------------------------------------------+

SELECT city_name, output_value
   FROM cities_of_interest,
       TABLE(return_two_copies(city_name) OVER (PARTITION BY 1))
   ORDER BY city_name, output_value;
+-----------+-------------------------------------------------------+
| CITY_NAME | OUTPUT_VALUE                                          |
|-----------+-------------------------------------------------------|
| Kyoto     | Kyoto                                                 |
| Kyoto     | Kyoto                                                 |
| Toronto   | Toronto                                               |
| Toronto   | Toronto                                               |
| Warsaw    | Warsaw                                                |
| Warsaw    | Warsaw                                                |
| NULL      | Created in constructor and output from endPartition() |
+-----------+-------------------------------------------------------+

Processamento de entradas muito grandes (por exemplo, arquivos grandes)¶

Em alguns casos, uma UDTF requer uma quantidade muito grande de memória para processar cada linha de entrada. Por exemplo, uma UDTF pode ler e processar um arquivo que é grande demais para caber na memória.

Para processar arquivos grandes em uma UDF ou UDTF, use a classe SnowflakeFile ou InputStream. Para obter mais informações, consulte Processamento de dados não estruturados usando UDFs ou UDTFs de Java.

Como usar uma tabela ou UDTF como entrada para uma UDTF¶

A entrada para uma função de tabela pode vir de uma tabela ou de outra UDTF, como documentado em Como usar uma tabela como entrada para uma função de tabela.

O exemplo abaixo mostra como utilizar uma tabela para fornecer entrada para a UDTF de Java split_file_into_words:

create table file_names (file_name varchar);
insert into file_names (file_name) values ('sample.txt'),
                                          ('sample_2.txt');

select f.file_name, w.word
   from file_names as f, table(split_file_into_words(f.file_name)) as w;

A saída se assemelha ao seguinte:

+-------------------+------------+
| FILE_NAME         | WORD       |
+-------------------+------------+
| sample_data.txt   | some       |
| sample_data.txt   | words      |
| sample_data_2.txt | additional |
| sample_data_2.txt | words      |
+-------------------+------------+

A cláusula imports da UDTF do Java deve especificar o nome e o caminho de cada arquivo passado para a UDTF. Por exemplo:

create function split_file_into_words(inputFileName string)
    ...
    imports = ('@inline_jars/sample.txt', '@inline_jars/sample_2.txt')
    ...

Cada arquivo já deve ter sido copiado para um estágio (neste caso, o estágio denominado @inline_jars) antes que a UDTF leia o arquivo.

Para obter um exemplo de uso de uma UDTF como entrada para outra UDTF, consulte Exemplos estendidos usando valores de tabela e outras UDTFs como entrada na documentação UDTF de JavaScript.