MaxCompute:Função com valor de tabela definida pelo usuário (UDTF) em Python 3

Ativar o Python 3

Por padrão, os projetos do MaxCompute usam o Python 2 para UDFs. Para usar o Python 3, adicione o seguinte comando no nível de sessão antes da instrução SQL e envie-os juntos:

set odps.sql.python.version=cp37;

Estrutura do código da UDTF

Use o MaxCompute Studio para escrever o código da UDTF em Python 3. Uma UDTF tem quatro componentes:

# Import the function signature module and the base class.
from odps.udf import annotate
from odps.udf import BaseUDTF

# Function signature: takes a STRING, returns a STRING.
@annotate('string -> string')

# Custom Python class derived from BaseUDTF.
class Explode(BaseUDTF):

    def process(self, arg):
        props = arg.split(',')
        for p in props:
            self.forward(p)

Limitações

O Python 3 não é compatível com o Python 2. Não é possível misturar UDTFs em Python 2 e Python 3 na mesma instrução SQL.

Migrar UDTFs em Python 2

O Python 2 atingiu o EOL. Migre as UDTFs existentes em Python 2 conforme a situação do projeto:

• Projeto novo ou primeira UDTF em Python: Escreva todas as UDTFs em Python diretamente na versão 3 desde o início.

• Projeto existente com muitas UDTFs em Python 2: Migre gradualmente para evitar interrupções. Escolha uma das seguintes abordagens:

  • Crie novas UDTFs em Python 3 e ative o Python 3 no nível de sessão para jobs que usem essas novas UDTFs. Para mais detalhes, consulte Ativar o Python 3.

  • Reescreva as UDTFs existentes em Python 2 para garantir compatibilidade com Python 2 e Python 3. Consulte Porting Python 2 Code to Python 3 para obter orientações.

Se uma UDTF for compartilhada entre vários projetos do MaxCompute, torne-a compatível com Python 2 e Python 3 para evitar falhas em projetos que ainda usam o Python 2.

Bibliotecas de terceiros

O NumPy não está incluído no ambiente de execução do Python 3 do MaxCompute. Para usar o NumPy em uma UDTF, carregue manualmente um pacote wheel do NumPy como recurso. O nome de arquivo esperado do Python Package Index (PyPI) ou de uma imagem é:

numpy-<Version>-cp37-cp37m-manylinux1_x86_64.whl

Para obter instruções sobre como carregar o pacote, consulte Operações de recursos ou Referenciar pacotes de terceiros em UDFs Python.

Assinaturas de função e tipos de dados

A assinatura de função declara os tipos de dados dos parâmetros de entrada e dos valores de retorno de uma UDTF. O MaxCompute valida a assinatura durante a análise semântica e retorna um erro caso os tipos reais não correspondam.

@annotate('arg_type_list -> type_list')

Referenciar recursos

Referencie arquivos e tabelas em uma UDTF Python usando o módulo odps.distcache.

odps.distcache.get_cache_file(resource_name)

Retorna o conteúdo de um recurso de arquivo.

• resource_name: nome de um recurso de arquivo existente no projeto MaxCompute. Retorna um erro se o nome for inválido ou se o arquivo não existir.

• Retorna um objeto semelhante a um arquivo. Chame close() no objeto após concluir a operação para liberar o identificador de arquivo.

• Declare o recurso de arquivo ao criar a UDTF. A omissão dessa declaração resulta em erro ao chamar a UDTF.

odps.distcache.get_cache_table(resource_name)

Retorna o conteúdo de um recurso de tabela.

• resource_name: nome de um recurso de tabela existente no projeto MaxCompute. Retorna um erro se o nome for inválido ou se a tabela não existir.

• Retorna um gerador. A iteração sobre ele produz um registro por linha, em que cada registro é um ARRAY.

O exemplo a seguir lê dados de um arquivo JSON e de um recurso de tabela e gera linhas com base em uma consulta:

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()

        # Append records from the table resource into the dict.
        records = list(get_cache_table('table_resource1'))
        for record in records:
            self.my_dict[record[0]] = record[1]

    def process(self, pageid):
        # For each input pageid, forward all associated adid values.
        for adid in self.my_dict[pageid]:
            self.forward(pageid, adid)

Chamar uma UDTF em Python 3

Após desenvolver uma UDTF em Python 3 seguindo o processo de desenvolvimento, chame-a no MaxCompute SQL.

• Dentro de um projeto: Chame a UDTF da mesma forma que uma função integrada.

• Entre projetos: Referencie uma UDTF de outro projeto usando o nome do projeto como prefixo:

  SELECT B:udf_in_other_project(arg0, arg1) AS res FROM table_t;

  Para configurar o compartilhamento de recursos entre projetos, consulte Compartilhamento de recursos entre projetos baseado em pacotes.

Para desenvolver e testar uma UDTF em Python 3 no MaxCompute Studio, consulte Desenvolver uma UDF Python.