How to write a UDTF in Python 2 - MaxCompute - Centro de documentación de Alibaba Cloud

MaxCompute ejecuta las funciones con valores de tabla definidas por el usuario (UDTF) de Python 2 mediante Python 2.7. Una UDTF recibe una fila de entrada y devuelve cero o más filas de salida. Esto resulta útil para operaciones como dividir o expandir datos.

Para crear y usar una UDTF de Python 2:

Escriba una clase de Python que extienda BaseUDTF e implemente los métodos obligatorios.
Registre la clase como una UDTF en MaxCompute y luego llámela desde MaxCompute SQL.

Estructura del código de una UDTF

Una UDTF de Python 2 se compone de hasta cinco elementos.

Componente	Obligatorio	Descripción
Declaración de codificación	No	Declara la codificación del archivo. Use `#coding:utf-8` o `# -- coding: utf-8 --`. Agréguela si el código contiene caracteres chinos; de lo contrario, MaxCompute devuelve un error durante la ejecución.
Importaciones de módulos	Sí	Debe incluir `from odps.udf import annotate` y `from odps.udf import BaseUDTF`. Agregue `from odps.distcache import get_cache_file` o `from odps.distcache import get_cache_table` si la UDTF hace referencia a recursos de archivos o tablas.
Firma de la función	No	Anota la UDTF con `@annotate(<signature>)` para declarar los tipos de datos de entrada y salida. Sin una firma, MaxCompute acepta cualquier tipo de entrada, pero trata todos los valores de salida como STRING.
Clase derivada	Sí	Una clase de Python que extiende `BaseUDTF`. Esta clase contiene toda la lógica de la UDTF.
Métodos de la clase	Sí	Implemente al menos `process`. Consulte la tabla de métodos a continuación.

Métodos

Método	Obligatorio	Descripción
`BaseUDTF.init()`	No	Inicializa el estado antes de procesar el primer registro. Si sobrescribe `init`, llame a `super(BaseUDTF, self).init()` al inicio. Úselo para configurar cualquier estado que la UDTF necesite mantener entre registros.
`BaseUDTF.process([args, ...])`	Sí	Se llama una vez por cada fila de entrada. Los argumentos coinciden con los parámetros de entrada de la UDTF declarados en SQL.
`BaseUDTF.forward([args, ...])`	Sí (se llama dentro de `process`)	Emite una fila de salida cada vez que se invoca. Llámelo una vez por cada fila que desee devolver. Si no se define ninguna firma de función, convierta todos los argumentos a STRING antes de llamar a `forward`.
`BaseUDTF.close()`	No	Se llama una vez antes de procesar el último registro. Úselo para liberar recursos o vaciar la salida.

Ejemplo

La siguiente UDTF divide una cadena separada por comas y emite cada valor como una fila independiente.

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

Firmas de función y tipos de datos

Formato de la firma

@annotate('arg_type_list -> type_list')

arg_type_list: lista de tipos de parámetros de entrada separados por comas. Use * para aceptar cualquier cantidad de argumentos, o déjelo vacío para no aceptar ninguno.
type_list: lista de tipos de valores de retorno separados por comas. Una UDTF puede devolver varias columnas.

A continuación se muestran ejemplos válidos de firmas.

Firma	Tipos de entrada	Tipos de retorno
`@annotate('bigint,boolean->string,datetime')`	BIGINT, BOOLEAN	STRING, DATETIME
`@annotate('*->string,datetime')`	Cualquier cantidad de argumentos	STRING, DATETIME
`@annotate('->double,bigint,string')`	Ninguna	DOUBLE, BIGINT, STRING
`@annotate("array<string>,struct<a1:bigint,b1:string>,string->map<string,bigint>,struct<b1:bigint>")`	ARRAY, STRUCT, STRING	MAP, STRUCT

Durante el análisis sintáctico semántico, MaxCompute comprueba que los tipos de datos de los argumentos reales coincidan con la firma. Si no coinciden, devuelve un error.

Nota

Los tipos de datos disponibles dependen de la edición de tipos de datos de su proyecto de MaxCompute. Para obtener más información, consulte Data type editions.

Asignaciones de tipos de datos

Escriba el código de la UDTF con los tipos de Python que corresponden a los tipos de MaxCompute SQL.

Tipo de MaxCompute SQL	Tipo de Python 2
BIGINT	int
STRING	str
DOUBLE	float
BOOLEAN	bool
DATETIME	int (milisegundos desde el 1 de enero 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

Notas adicionales sobre el manejo de tipos:

NULL en MaxCompute SQL corresponde a None en Python.
odps.udf.int(value, silent=True) devuelve None en lugar de generar un error cuando el valor no se puede convertir a int.

Referencia a recursos de archivos y tablas

Use el módulo odps.distcache para cargar recursos de archivos o tablas en su UDTF.

get_cache_file(resource_name): devuelve un objeto similar a un archivo para el recurso de archivo especificado. Llame a close() sobre el objeto cuando termine. Declare el recurso de archivo al registrar la UDTF; de lo contrario, la llamada fallará durante la ejecución.
get_cache_table(resource_name): devuelve un generador sobre el recurso de tabla especificado. Cada iteración produce un registro como una lista (tipo ARRAY).

El siguiente ejemplo carga un archivo JSON y un recurso de tabla, y luego los usa para buscar identificadores de anuncios por identificador 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)

Llamada a la UDTF en MaxCompute SQL

Después de completar development process, llame a la UDTF desde MaxCompute SQL:

Dentro de un proyecto: llame a la UDTF de la misma manera que llama a las funciones integradas.
Entre proyectos: para usar una UDTF del proyecto B en el proyecto A, anteponga el nombre del proyecto al nombre de la función:
```
SELECT B:udf_in_other_project(arg0, arg1) AS res FROM table_t;
```
Para obtener más información, consulte Acceso a recursos entre proyectos basado en paquetes.

Limitaciones

MaxCompute ejecuta el código de las UDTF de Python 2 en un entorno sandbox. No se permiten las siguientes operaciones:

Leer o escribir archivos locales
Iniciar subprocesos
Iniciar hilos
Abrir conexiones de socket
Llamar a UDF de Python 2 desde otros sistemas

Cargue únicamente código que use bibliotecas estándar de Python. Los módulos o módulos de extensión en C que dependan de las operaciones restringidas anteriores no están disponibles.

Módulos de extensión en C disponibles

Los siguientes módulos de extensión en C están disponibles en el 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

También están disponibles todos los módulos implementados completamente en Python que no dependan de módulos de extensión.

Límite de tamaño de salida

La escritura en sys.stdout o sys.stderr tiene un límite de 20 KB. Los caracteres que superen este límite se descartan sin notificación.

Bibliotecas de terceros

Las bibliotecas de terceros, como NumPy, vienen preinstaladas en el entorno de Python 2 de MaxCompute. El acceso a datos locales y la mayoría de las API de E/S de red están deshabilitadas para estas bibliotecas; solo se admite una E/S de red limitada.

MaxCompute:Python 2 UDTF