How to write a UDTF in Python 2 - MaxCompute: Central de Documentação do Alibaba Cloud

O MaxCompute executa funções de tabela definidas pelo usuário (UDTFs) em Python 2 com a versão 2.7. Uma UDTF recebe uma linha de entrada e retorna zero ou mais linhas de saída, o que é útil para operações como divisão ou expansão de dados.

Para criar e usar uma UDTF em Python 2:

Escreva uma classe Python que estenda BaseUDTF e implemente os métodos obrigatórios.
Registre a classe como UDTF no MaxCompute e chame-a no MaxCompute SQL.

Estrutura do código da UDTF

Uma UDTF em Python 2 tem até cinco componentes.

Componente	Obrigatório	Descrição
Declaração de codificação	Não	Define a codificação do arquivo. Use `#coding:utf-8` ou `# -- coding: utf-8 --`. Adicione esta declaração se o código contiver caracteres chineses. Sem ela, o MaxCompute retorna um erro durante a execução.
Importação de módulos	Sim	Inclua `from odps.udf import annotate` e `from odps.udf import BaseUDTF`. Adicione `from odps.distcache import get_cache_file` ou `from odps.distcache import get_cache_table` se a UDTF referenciar recursos de arquivo ou tabela.
Assinatura da função	Não	Anota a UDTF com `@annotate(<signature>)` para declarar os tipos de dados de entrada e saída. Sem assinatura, o MaxCompute aceita qualquer tipo de entrada, mas trata todos os valores de saída como STRING.
Classe derivada	Sim	Classe Python que estende `BaseUDTF`. Esta classe contém toda a lógica da UDTF.
Métodos da classe	Sim	Implemente pelo menos o método `process`. Consulte a tabela de métodos abaixo.

Métodos

Método	Obrigatório	Descrição
`BaseUDTF.init()`	Não	Inicializa o estado antes de processar o primeiro registro. Se você substituir `init`, chame `super(BaseUDTF, self).init()` no início. Use este método para configurar qualquer estado que a UDTF precise manter entre registros.
`BaseUDTF.process([args, ...])`	Sim	O sistema chama este método uma vez para cada linha de entrada. Os argumentos correspondem aos parâmetros de entrada da UDTF declarados no SQL.
`BaseUDTF.forward([args, ...])`	Sim (chamado dentro de `process`)	Emite uma linha de saída a cada chamada. Chame-o uma vez para cada linha que desejar retornar. Se não houver assinatura de função definida, converta todos os argumentos para STRING antes de chamar `forward`.
`BaseUDTF.close()`	Não	O sistema executa este método uma vez antes de processar o último registro. Use-o para liberar recursos ou descarregar a saída.

Exemplo

A UDTF a seguir divide uma string separada por vírgulas e emite cada valor como uma linha distinta.

#coding:utf-8
from odps.udf import annotate
from odps.udf import BaseUDTF

@annotate('string -> string')
class Explode(BaseUDTF):
    def process(self, arg):
        props = arg.split(',')
        for p in props:
            self.forward(p)

Assinaturas de função e tipos de dados

Formato da assinatura

@annotate('arg_type_list -> type_list')

arg_type_list: lista de tipos de parâmetros de entrada separados por vírgula. Use * para aceitar qualquer número de argumentos ou deixe em branco para não aceitar argumentos.
type_list: lista de tipos de valores de retorno separados por vírgula. Uma UDTF pode retornar várias colunas.

A tabela a seguir mostra exemplos válidos de assinaturas.

Assinatura	Tipos de entrada	Tipos de retorno
`@annotate('bigint,boolean->string,datetime')`	BIGINT, BOOLEAN	STRING, DATETIME
`@annotate('*->string,datetime')`	Qualquer número de argumentos	STRING, DATETIME
`@annotate('->double,bigint,string')`	Nenhum	DOUBLE, BIGINT, STRING
`@annotate("array<string>,struct<a1:bigint,b1:string>,string->map<string,bigint>,struct<b1:bigint>")`	ARRAY, STRUCT, STRING	MAP, STRUCT

Durante a análise semântica, o MaxCompute verifica se os tipos de dados dos argumentos reais correspondem à assinatura. Incompatibilidades geram erro.

Nota

Os tipos de dados disponíveis dependem da edição de tipos de dados do seu projeto MaxCompute. Para obter mais informações, consulte Edições de tipos de dados.

Mapeamento de tipos de dados

Escreva o código da UDTF com os tipos Python correspondentes aos tipos do MaxCompute SQL.

Tipo MaxCompute SQL	Tipo Python 2
BIGINT	int
STRING	str
DOUBLE	float
BOOLEAN	bool
DATETIME	int (milissegundos desde 1º de janeiro de 1970, 00:00:00 UTC)
FLOAT	float
CHAR	str
VARCHAR	str
BINARY	bytearray
DATE	int
DECIMAL	decimal.Decimal
ARRAY	list
MAP	dict
STRUCT	collections.namedtuple

Observações adicionais sobre o tratamento de tipos:

O valor NULL no MaxCompute SQL corresponde a None em Python.
A função odps.udf.int(value, silent=True) retorna None em vez de gerar erro quando não é possível converter o valor para int.

Referenciar recursos de arquivo e tabela

Use o módulo odps.distcache para carregar recursos de arquivo ou tabela na UDTF.

get_cache_file(resource_name): retorna um objeto semelhante a arquivo para o recurso especificado. Chame close() no objeto após concluir a operação. Declare o recurso de arquivo ao registrar a UDTF. Caso contrário, a chamada falhará durante a execução.
get_cache_table(resource_name): retorna um gerador para o recurso de tabela especificado. Cada iteração produz um registro como lista (tipo ARRAY).

O exemplo a seguir carrega um arquivo JSON e um recurso de tabela e os usa para buscar IDs de anúncio por ID de página.

# -*- coding: utf-8 -*-
from odps.udf import annotate
from odps.udf import BaseUDTF
from odps.distcache import get_cache_file
from odps.distcache import get_cache_table

@annotate('string -> string, bigint')
class UDTFExample(BaseUDTF):
    def __init__(self):
        import json
        # Load the JSON file resource into a dict
        cache_file = get_cache_file('test_json.txt')
        self.my_dict = json.load(cache_file)
        cache_file.close()
        # Merge records from the table resource
        records = list(get_cache_table('table_resource1'))
        for record in records:
            self.my_dict[record[0]] = [record[1]]

    def process(self, pageid):
        # Emit one row per ad ID associated with the page
        for adid in self.my_dict[pageid]:
            self.forward(pageid, adid)

Chamar a UDTF no MaxCompute SQL

Após concluir o processo de desenvolvimento, chame a UDTF no MaxCompute SQL:

No mesmo projeto: chame a UDTF como você chama as funções integradas.
Entre projetos: para usar uma UDTF do projeto B no projeto A, prefixe o nome da função com o nome do projeto:
```
SELECT B:udf_in_other_project(arg0, arg1) AS res FROM table_t;
```
Para obter mais informações, consulte Acesso a recursos entre projetos baseado em pacotes.

Limitações

O MaxCompute executa o código da UDTF em Python 2 em um ambiente sandbox. As seguintes operações não são permitidas:

Ler ou gravar arquivos locais
Iniciar subprocessos
Iniciar threads
Abrir conexões de socket
Chamar UDFs Python 2 de outros sistemas

Faça upload apenas de código que use bibliotecas padrão do Python. Módulos ou extensões C que dependam das operações restritas acima não estão disponíveis.

Módulos de extensão C disponíveis

Os seguintes módulos de extensão C estão disponíveis no sandbox:

array, audioop, binascii, bisect, cmath, _codecs_cn, _codecs_hk, _codecs_iso2022, _codecs_jp, _codecs_kr, _codecs_tw, _collections, cStringIO, datetime, _functools, future_builtins, _heapq, _hashlib, itertools, _json, _locale, _lsprof, math, _md5, _multibytecodec, operator, _random, _sha256, _sha512, _sha, _struct, strop, time, unicodedata, _weakref, cPickle

Todos os módulos implementados puramente em Python que não dependem de módulos de extensão também estão disponíveis.

Limite de tamanho de saída

A gravação em sys.stdout ou sys.stderr tem limite de 20 KB. O sistema descarta silenciosamente os caracteres que excedem esse limite.

Bibliotecas de terceiros

Bibliotecas de terceiros, como NumPy, vêm pré-instaladas no ambiente Python 2 do MaxCompute. O acesso a dados locais e a maioria das APIs de E/S de rede ficam desativados para bibliotecas de terceiros. Apenas E/S de rede limitada está disponível.

MaxCompute:Python 2 UDTF