MaxCompute:UDAF em Python 3

Use o MaxCompute Studio para escrever o código de uma função de agregação definida pelo usuário (UDAF) em Python 3. O código deve conter os seguintes elementos:

• Importação de módulos: Obrigatório.

  Importe pelo menos from odps.udf import annotate e from odps.udf import BaseUDAF. A instrução from odps.udf import annotate importa o módulo de assinatura da função, permitindo que o MaxCompute reconheça a assinatura definida no código. A instrução from odps.udf import BaseUDAF importa a classe base para UDAFs em Python. Implemente métodos como iterate, merge e terminate na classe derivada.

  Se o código da UDAF precisar referenciar recursos de arquivo ou tabela, inclua from odps.distcache import get_cache_file para recursos de arquivo ou from odps.distcache import get_cache_table para recursos de tabela.

• Assinatura da função: Obrigatório.

  O formato é @annotate(<signature>). O parâmetro signature define os tipos de dados dos parâmetros de entrada e do valor de retorno da função. Para obter mais informações sobre assinaturas de função, consulte Function signature and data types.

• Classe Python personalizada (classe derivada): Obrigatório.

  Esta classe atua como unidade organizacional do código da UDAF e define as variáveis e os métodos que implementam a lógica de negócios. Também é possível referenciar bibliotecas de terceiros integradas, arquivos ou recursos de tabela no código. Para mais detalhes, consulte Third-party libraries ou Reference resources.

• Implementação dos métodos da classe Python: Obrigatório.

  A implementação da classe Python inclui os métodos listados abaixo. Implemente-os conforme a necessidade.

  Definição do método	Descrição
  BaseUDAF.new_buffer()	Retorna um buffer para o valor intermediário da função de agregação. O buffer deve ser um objeto Marshal, como LIST ou DICT. O tamanho do buffer não deve aumentar proporcionalmente ao volume de dados. Em casos extremos, o tamanho do buffer após a serialização do objeto não pode exceder 2 MB.
  BaseUDAF.iterate(buffer[, args, ...])	Agrega os argumentos args ao valor intermediário armazenado em buffer.
  BaseUDAF.merge(buffer, pbuffer)	Combina os valores intermediários buffer e pbuffer, armazenando o resultado em buffer.
  BaseUDAF.terminate(buffer)	Converte o conteúdo de buffer em um tipo de dados primitivo do MaxCompute SQL.

O código a seguir apresenta um exemplo de UDAF.

# Import the function signature module and the base class.
from odps.udf import annotate
from odps.udf import BaseUDAF
# Function signature.
@annotate('double->double')
# Custom Python class.
class Average(BaseUDAF):
# Implement the methods of the Python class.
    def new_buffer(self):
        return [0, 0]
    def iterate(self, buffer, number):
        if number is not None:
            buffer[0] += number
            buffer[1] += 1
    def merge(self, buffer, pbuffer):
        buffer[0] += pbuffer[0]
        buffer[1] += pbuffer[1]
    def terminate(self, buffer):
        if buffer[1] == 0:
            return 0.0
        return buffer[0] / buffer[1]

A figura a seguir ilustra a lógica de implementação e o fluxo de cálculo de uma UDAF do MaxCompute para calcular o valor médio (

avg

).

O parâmetro

pbuffer

corresponde a

pr

na figura, enquanto

buffer

corresponde a

r

.

Como a Python Software Foundation encerrará em breve a manutenção do Python 2, recomendamos migrar seus projetos conforme o tipo:

• Novos projetos: Aplica-se a novos projetos do MaxCompute ou a projetos nos quais você está escrevendo uma UDAF em Python pela primeira vez. Recomendamos escrever todas as UDAFs em Python usando a versão 3.

• Projetos existentes: Aplica-se a projetos do MaxCompute que já possuem muitas UDAFs em Python 2. Ative o Python 3 com cautela. Se planeja migrar gradualmente todas as UDAFs do Python 2 para o Python 3, utilize os métodos a seguir:

  • Novos jobs e novas UDAFs: Escreva-os em Python 3 e ative o Python 3 no nível da sessão. Para obter mais informações sobre como ativar o Python 3, consulte Enable Python 3.

  • UDAFs em Python 2: Reescreva o código das UDAFs existentes para garantir compatibilidade tanto com Python 2 quanto com Python 3. Para saber mais sobre como reescrever o código, consulte Porting Python 2 Code to Python 3.

    Nota

    Se você pretende criar uma UDAF pública e conceder permissões de uso para vários projetos do MaxCompute, recomendamos tornar a UDAF compatível com Python 2 e Python 3 simultaneamente.

set odps.sql.python.version=cp37;

A assinatura da função segue o formato abaixo.

@annotate(<signature>)

O parâmetro

signature

é uma string que identifica os tipos de dados dos parâmetros de entrada e do valor de retorno. Ao executar uma UDAF, os tipos de dados de entrada e saída devem corresponder exatamente aos tipos especificados na assinatura. Durante a fase de análise da consulta, o sistema valida a chamada da função comparando-a com a assinatura definida. Caso haja incompatibilidade de tipos, um erro será reportado. O formato específico é o seguinte:

'arg_type_list -> type'

onde:

• arg_type_list: Representa os tipos de dados dos parâmetros de entrada. É possível especificar múltiplos parâmetros separados por vírgulas (,). Os tipos suportados incluem BIGINT, STRING, DOUBLE, BOOLEAN, DATETIME, DECIMAL, FLOAT, BINARY, DATE, DECIMAL(precision,scale), CHAR, VARCHAR, tipos complexos (ARRAY, MAP, STRUCT) e tipos complexos aninhados.

  O campo arg_type_list também aceita um asterisco (*) ou uma string vazia ('').

  • Quando arg_type_list for um asterisco (*), indica que a função aceita qualquer quantidade de parâmetros de entrada.

  • Quando arg_type_list for uma string vazia (''), indica que a função não possui parâmetros de entrada.

  Para mais detalhes sobre a sintaxe estendida da anotação Resolve, consulte Dynamic parameters for UDAFs and UDTFs.

• type: Representa o tipo de dado do valor de retorno. Uma UDAF retorna apenas uma coluna. Os tipos suportados incluem BIGINT, STRING, DOUBLE, BOOLEAN, DATETIME, DECIMAL, FLOAT, BINARY, DATE, DECIMAL(precision,scale), tipos complexos (ARRAY, MAP, STRUCT) e tipos complexos aninhados.

Após desenvolver uma UDAF em Python 3 seguindo as instruções em development flow, chame-a em uma instrução SQL do MaxCompute. Os métodos de chamada são os seguintes:

• Usar uma UDF dentro de um projeto do MaxCompute: O método é semelhante ao uso de built-in functions. Utilize a função definida pelo usuário da mesma forma que usaria uma função integrada.

• Usar uma UDF entre projetos: Utilize uma UDF do Projeto B dentro do Projeto A. A instrução a seguir mostra um exemplo: select B:udf_in_other_project(arg0, arg1) as res from table_t;. Para obter mais informações sobre compartilhamento entre projetos, consulte Cross-project resource access based on packages.

Assinatura da função

Exemplo de UDAF

from odps.udf import annotate
from odps.udf import BaseUDAF
@annotate('bigint,*->string')
class MultiColSum(BaseUDAF):
    def new_buffer(self):
        return [0]
    def iterate(self, buffer, *args):
        for arg in args:
            buffer[0] += int(arg)
    def merge(self, buffer, pbuffer):
        buffer[0] += pbuffer[0]
    def terminate(self, buffer):
        return str(buffer[0])

-- Sums multiple input parameters.
SELECT my_multi_col_sum(a,b,c,d,e) from values (1,"2","3","4","5"), (6,"7","8","9","10") t(a,b,c,d,e);
-- The return value is 55.