Como ler recursos do MaxCompute usando uma UDTF Python 3 - MaxCompute: Central de Documentação do Alibaba Cloud

Utilize uma função de tabela definida pelo usuário (UDTF) em Python 3 para ler recursos do MaxCompute — incluindo arquivos e tabelas de recursos — no cliente MaxCompute.

Pré-requisitos

Antes de começar, verifique se você possui:

O cliente MaxCompute instalado. Para mais informações, consulte Instale e configure o cliente MaxCompute.

Métodos da classe handler

Uma UDTF Python 3 é implementada como uma classe que estende BaseUDTF. Os métodos da classe são:

Método	Obrigatório	Descrição
`__init__`	Não	Inicializa o estado. Utilize este método para carregar arquivos e tabelas de recursos uma única vez na inicialização, antes do processamento de qualquer linha.
`process`	Sim	Processa cada linha de entrada. Chame `forward()` dentro deste método para emitir as linhas de saída.

O decorador @annotate define a assinatura da função. Chame forward(*args) dentro de process para emitir cada linha de saída — uma chamada a forward produz uma linha de saída.

Parâmetros dinâmicos

Para obter uma referência completa sobre formatos de assinatura de função e tipos de dados, consulte Assinaturas de função e tipos de dados.

Os asteriscos (*) nas assinaturas @annotate possuem significados diferentes dependendo da posição:

Em listas de parâmetros: * indica que a entrada aceita qualquer quantidade de parâmetros adicionais de qualquer tipo. Por exemplo, @annotate('double,*->string') declara um primeiro parâmetro DOUBLE seguido por um número variável de parâmetros. É necessário compilar o código para calcular a quantidade e os tipos dos parâmetros de entrada, gerenciando-os com base na função printf da linguagem C. O método process recebe esses valores como *args e deve tratá-los explicitamente.

Em valores de retorno: * significa que será retornado um número variável de valores STRING, cuja quantidade é determinada pelo número de aliases fornecidos no momento da chamada. Por exemplo, @annotate("bigint,string->double,*") chamado como UDTF(x, y) as (a, b, c) retorna três valores: a como DOUBLE, e b e c como STRING. A chamada forward dentro de process deve emitir um array cujo comprimento corresponda exatamente à contagem de aliases — uma incompatibilidade causa um erro em tempo de execução, não em tempo de compilação.

Nota

Esta regra do * nos valores de retorno aplica-se apenas a UDTFs. Funções agregadas definidas pelo usuário (UDAFs) sempre retornam exatamente um valor.

Código de exemplo

Ler recursos do MaxCompute

A UDTF a seguir lê mapeamentos de página para anúncio a partir de um arquivo de recurso JSON (test_json.txt) e de uma tabela de recursos (table_resource1), emitindo em seguida uma linha por ID de anúncio para cada ID de página de entrada.

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):
    """Read pageid and adid_list from the file get_cache_file and the table get_cache_table to generate dict.
    """
    def __init__(self):
        import json
        cache_file = get_cache_file('test_json.txt')
        self.my_dict = json.load(cache_file)
        cache_file.close()
        records = list(get_cache_table('table_resource1'))
        for record in records:
            self.my_dict[record[0]] = record[1]
    """Enter pageid and generate pageid and all adid values.
    """
    def process(self, pageid):
        for adid in self.my_dict[pageid]:
            self.forward(pageid, adid)

As funções get_cache_file e get_cache_table (do módulo odps.distcache) carregam recursos pelo nome utilizado quando o recurso foi registrado com add file ou add table. Os recursos são carregados uma única vez em __init__ e reutilizados em todas as chamadas de process.

Usar parâmetros dinâmicos

A UDTF abaixo analisa uma string JSON e extrai valores por chave. O número de valores retornados é igual ao número de parâmetros de entrada.

from odps.udf import annotate
from odps.udf import BaseUDTF
import json

@annotate('string,*->string,*')
class JsonTuple(BaseUDTF):
    def process(self, *args):
        length = len(args)
        result = [None] * length
        try:
            obj = json.loads(args[0])
            for i in range(1, length):
                result[i] = str(obj.get(args[i]))
        except Exception as err:
            result[0] = str(err)
            for i in range(1, length):
                result[i] = None
        self.forward(*result)

O primeiro argumento (args[0]) é a string JSON. Os argumentos adicionais são as chaves a serem extraídas. O primeiro valor de retorno contém qualquer erro de análise; os valores subsequentes contêm o conteúdo extraído na ordem das chaves.

Chame esta UDTF com um número correspondente de aliases:

-- Number of output aliases matches number of input parameters
SELECT my_json_tuple(json, 'a', 'b') as (exceptions, a, b) FROM jsons;

-- The variable-length part can have no columns
SELECT my_json_tuple(json) as exceptions FROM jsons;

-- This causes a runtime error: alias count (4) does not match input parameter count (3)
SELECT my_json_tuple(json, 'a', 'b') as (exceptions, a, b, c) FROM jsons;

Registrar e chamar uma UDTF

O procedimento a seguir utiliza UDTFExample para demonstrar o fluxo de trabalho completo. Salve o código como py_udtf_example.py na pasta bin do cliente MaxCompute antes de iniciar.

Etapa 1: Crie tabelas de recursos e preparar dados

Faça login no cliente MaxCompute. Para mais informações, consulte Iniciar o cliente MaxCompute.

Crie e popule a tabela de recursos table_resource1:

create table if not exists table_resource1 (pageid string, adid_list array<int>);
insert into table table_resource1 values("contact_page2",array(2,3,4)),("contact_page3",array(5,6,7));

Nota

O campo adid_list é do tipo ARRAY. Para permitir que o Python 3 leia dados do tipo ARRAY, execute set odps.sql.python.version=cp37; no nível da sessão antes de realizar consultas.

Crie e popule a tabela interna tmp1:

create table if not exists tmp1 (pageid string);
insert into table tmp1 values ("front_page"),("contact_page1"),("contact_page3");

Coloque test_json.txt na pasta bin do cliente MaxCompute. O arquivo contém:

{"front_page":[1, 2, 3], "contact_page1":[3, 4, 5]}

Etapa 2: Registrar recursos

Adicione o arquivo Python, o arquivo JSON e a tabela como recursos do MaxCompute. Para mais informações, consulte Adicionar recursos.

add py py_udtf_example.py;
add file test_json.txt;
add table table_resource1 as table_resource1;

Etapa 3: Crie a UDTF

Registre a UDTF, listando todos os recursos dos quais ela depende. Para mais informações, consulte Crie uma UDF.

create function my_udtf as 'py_udtf_example.UDTFExample' using 'py_udtf_example.py, test_json.txt, table_resource1';

Etapa 4: Chamar a UDTF

Três padrões de chamada são suportados. Todos produzem a mesma saída para este exemplo.

Chamada direta:

select my_udtf(pageid) as (pageid, adid) from tmp1;

Com LATERAL VIEW:

select pageid, adid from tmp1 lateral view my_udtf(pageid) adTable as udtf_pageid, adid;

Com LATERAL VIEW e uma função agregada:

select adid, count(1) as cnt
    from tmp1 lateral view my_udtf(pageid) adTable as udtf_pageid, adid
group by adid;

As consultas de chamada direta e com LATERAL VIEW retornam as mesmas 9 linhas:

+------------+------------+
| pageid     | adid       |
+------------+------------+
| front_page | 1          |
| front_page | 2          |
| front_page | 3          |
| contact_page1 | 3       |
| contact_page1 | 4       |
| contact_page1 | 5       |
| contact_page3 | 5       |
| contact_page3 | 6       |
| contact_page3 | 7       |
+------------+------------+

A consulta agregada retorna:

+------------+------------+
| adid       | cnt        |
+------------+------------+
| 1          | 1          |
| 2          | 1          |
| 3          | 2          |
| 4          | 1          |
| 5          | 2          |
| 6          | 1          |
| 7          | 1          |
+------------+------------+