MaxCompute:Desarrollo de UDF (Python 3)

Estructura del código de una 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

Límites

• Acceso a Internet mediante UDF

  De forma predeterminada, MaxCompute no permite acceder a Internet mediante UDF. Si necesita este acceso, complete el formulario de solicitud de conexión de red según sus requisitos de negocio y envíe la solicitud. El equipo de soporte técnico de MaxCompute se pondrá en contacto con usted a la brevedad para habilitar la conectividad de red. Para obtener más información sobre cómo completar el formulario de solicitud de conexión de red, consulte Network connection process.

• Acceso a una VPC mediante UDF

  Por defecto, MaxCompute no permite acceder a recursos en VPC mediante UDF. Para lograrlo, debe establecer una conexión de red entre MaxCompute y la VPC. Para obtener más información sobre las operaciones relacionadas, consulte Access VPC resources from a UDF.

• Lectura de datos de tablas mediante UDF, UDAF o UDTF

  No es posible usar UDF, UDAF ni UDTF para leer datos de los siguientes tipos de tablas:

  • Tablas en las que se aplicó evolución de esquema

  • Tablas que contienen tipos de datos complejos

  • Tablas que contienen tipos de datos JSON

  • Tablas transaccionales

Notas de uso

Python 3 no es compatible con Python 2, por lo que no se pueden usar ambos en la misma instrucción SQL. Considere esta compatibilidad antes de cambiar de versión.

Desarrollo de UDF: flujo de trabajo general

El desarrollo de una UDF implica preparar el entorno, escribir el código, cargar y registrar la UDF y, finalmente, invocarla. En las siguientes secciones se detalla este flujo de trabajo con MaxCompute Studio, DataWorks y odpscmd.

Instalar la biblioteca NumPy

El entorno de ejecución integrado de Python 3 no incluye NumPy. Si su UDF requiere NumPy, cargue manualmente el paquete WHEEL. Al descargar el paquete desde PyPI o un sitio espejo, el nombre del archivo tiene el formato numpy-<version>-cp37-cp37m-manylinux1_x86_64.whl. Para obtener más información sobre cómo cargar un paquete, consulte Resource operations o Use a third-party package in a Python UDF.

Para consultar la lista de bibliotecas estándar compatibles con Python 3, visite biblioteca estándar de Python 3.

Firmas de función y tipos de datos

@annotate(<signature>)

'arg_type_list -> type'

La siguiente tabla describe las asignaciones entre los tipos de datos compatibles con MaxCompute SQL y los tipos de datos de Python 2. Debe escribir las UDF de Python basándose en estas asignaciones para garantizar la coherencia de los tipos de datos.

Referencia a recursos

Puede hacer referencia a archivos o tablas en el código de UDF de Python 2 mediante el módulo odps.distcache.

• odps.distcache.get_cache_file(resource_name, mode): devuelve el contenido de un recurso de archivo específico en el mode indicado.

  • resource_name es una cadena que especifica el nombre de una tabla existente en su proyecto de MaxCompute. Si el nombre de la tabla no es válido o la tabla no existe, se devuelve un error.

  • El parámetro mode es de tipo STRING. Su valor predeterminado es 't'. Si establece mode en 't', el archivo se abre en modo texto. Si establece mode en 'b', el archivo se abre en modo binario.

  • El valor de retorno es un objeto similar a un archivo. Si este objeto ya no se utiliza, debe invocar el método close para liberar el archivo abierto.

  El siguiente código muestra cómo hacer referencia a un archivo.

  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): devuelve el contenido de un recurso de tabla específico.

  • El parámetro resource_name especifica un recurso de tabla existente en el proyecto actual de MaxCompute. Se genera una excepción si el nombre del recurso no es válido o si el recurso no existe. Tipos de datos compatibles: BIGINT, STRING, DOUBLE, BOOLEAN, DATETIME, FLOAT, CHAR, VARCHAR, BINARY, DATE, DECIMAL, ARRAY, MAP y STRUCT.

  • El valor de retorno es un generador (Generator). Cada iteración produce un registro de la tabla en forma de arreglo.

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)

Invocación de UDF

set odps.sql.python.version=cp37;

Migrar UDF de Python 2

Ejemplos 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