MaxCompute:Desenvolvimento de UDF (Python 3)

Estrutura do código da UDF

# Import the function signature module.
from odps.udf import annotate
# Define the function signature.
@annotate("bigint,bigint->bigint")
# Define the custom Python class.
class MyPlus(object):
# Implement the evaluate method.
    def evaluate(self, arg0, arg1):
        if None in (arg0, arg1):
            return None
        return arg0 + arg1

Limites

• Acesso à Internet por meio de UDFs

  Por padrão, o MaxCompute não permite acesso à Internet por meio de UDFs. Se for necessário acessar a Internet usando UDFs, preencha o formulário de solicitação de conexão de rede conforme suas necessidades de negócios e envie a solicitação. A equipe de suporte técnico do MaxCompute entrará em contato prontamente para ative a conectividade de rede. Para mais informações sobre como preencher o formulário de solicitação de conexão de rede, consulte Network connection process.

• Acesso a uma VPC por meio de UDFs

  Por padrão, o MaxCompute não permite o acesso a recursos em VPCs usando UDFs. Para usar UDFs no acesso a recursos dentro de uma VPC, estabeleça uma conexão de rede entre o MaxCompute e a VPC. Para mais informações sobre as operações relacionadas, consulte Access VPC resources from a UDF.

• Leitura de dados de tabela via UDFs, UDAFs ou UDTFs

  Não é possível usar UDFs, UDAFs ou UDTFs para ler dados dos seguintes tipos de tabela:

  • Tabela com evolução de schema aplicada

  • Tabela com tipos de dados complexos

  • Tabela com tipos de dados JSON

  • Tabela transacional

Observações de uso

O Python 3 não é compatível com o Python 2, e ambos não podem ser usados na mesma instrução SQL. Considere a compatibilidade antes de migrar.

Desenvolvimento de UDF: fluxo de trabalho geral

O desenvolvimento de UDF envolve preparar o ambiente, escrever o código, fazer upload e registrar a UDF e, em seguida, chamá-la. As seções a seguir detalham esse fluxo de trabalho com MaxCompute Studio, DataWorks e odpscmd.

Instalar a biblioteca NumPy

O ambiente de execução integrado do Python 3 não inclui o NumPy. Se sua UDF exigir NumPy, faça upload manual do pacote WHEEL. Ao baixar o pacote do PyPI ou de um site espelho, o nome do arquivo terá o formato numpy-<version>-cp37-cp37m-manylinux1_x86_64.whl. Para mais informações sobre como fazer upload de um pacote, consulte Resource operations ou Use a third-party package in a Python UDF.

Para obter uma lista das bibliotecas padrão suportadas pelo Python 3, consulte Biblioteca padrão do Python 3.

Assinaturas de função e tipos de dados

@annotate(<signature>)

'arg_type_list -> type'

A tabela a seguir descreve os mapeamentos entre os tipos de dados suportados no MaxCompute SQL e os tipos de dados do Python 2. Escreva as UDFs Python com base nesses mapeamentos para garantir a consistência dos tipos de dados.

Referenciar recursos

É possível referenciar arquivos ou tabelas no código de UDF Python 2 usando o módulo odps.distcache.

• odps.distcache.get_cache_file(resource_name, mode): Retorna o conteúdo de um recurso de arquivo especificado no mode indicado.

  • resource_name é uma string que especifica o nome de uma tabela existente no seu projeto MaxCompute. Se o nome da tabela for inválido ou se a tabela não existir, um erro será retornado.

  • O parâmetro mode é uma STRING. O valor padrão é 't'. Se você definir mode como 't', o arquivo será aberto em modo texto. Se definir mode como 'b', o arquivo será aberto em modo binário.

  • O valor de retorno é um objeto semelhante a um arquivo. Se esse objeto não for mais utilizado, chame o método close para liberar o arquivo aberto.

  O código a seguir mostra como referenciar um arquivo.

  from odps.udf import annotate
  from odps.distcache import get_cache_file
  @annotate('bigint->string')
  class DistCacheExample(object):
  def __init__(self):
      cache_file = get_cache_file('test_distcache.txt')
      kv = {}
      for line in cache_file:
          line = line.strip()
          if not line:
              continue
          k, v = line.split()
          kv[int(k)] = v
      cache_file.close()
      self.kv = kv
  def evaluate(self, arg):
      return self.kv.get(arg)

• odps.distcache.get_cache_table(resource_name): Retorna o conteúdo de um recurso de tabela especificado.

  • O parâmetro resource_name especifica um recurso de tabela existente no projeto MaxCompute atual. Uma exceção será lançada se o nome do recurso for inválido ou se o recurso não existir. Tipos de dados suportados: BIGINT, STRING, DOUBLE, BOOLEAN, DATETIME, FLOAT, CHAR, VARCHAR, BINARY, DATE, DECIMAL, ARRAY, MAP e STRUCT.

  • O valor de retorno é um Generator. Cada iteração produz um registro da tabela na forma de array.

from odps.udf import annotate
from odps.distcache import get_cache_table
@annotate('->string')
class DistCacheTableExample(object):
    def __init__(self):
        self.records = list(get_cache_table('udf_test'))
        self.counter = 0
        self.ln = len(self.records)
    def evaluate(self):
        if self.counter > self.ln - 1:
            return None
        ret = self.records[self.counter]
        self.counter += 1
        return str(ret)

Chamar UDFs

set odps.sql.python.version=cp37;

Migrar UDFs Python 2

Exemplos de UDF

• Write a Hive UDF in Java

• Use complex data types in Java UDFs

• Replace strings by using regular expressions

• Obtain the values of strings that have no delimiters

• Obtain the values of strings that have delimiters

• Reference a file resource

• Reference a table resource

• Reference third-party packages in Python UDFs