Utilice una función con valor de tabla definida por el usuario (UDTF) de Python 3 para leer recursos de MaxCompute —incluidos archivos y tablas de recursos— desde el cliente de MaxCompute.
Requisitos previos
Antes de empezar, asegúrese de contar con lo siguiente:
El cliente de MaxCompute instalado. Para obtener más información, consulte Instalación y configuración del cliente de MaxCompute.
Métodos de la clase controladora
Una UDTF de Python 3 se implementa como una clase que extiende BaseUDTF. Los métodos de la clase son los siguientes:
|
Método |
Obligatorio |
Descripción |
|
|
No |
Inicializa el estado. Úselo para cargar archivos y tablas de recursos una sola vez al inicio, antes de procesar cualquier fila. |
|
|
Sí |
Procesa cada fila de entrada. Llame a |
El decorador @annotate define la firma de la función. Llame a forward(*args) dentro de process para emitir cada fila de salida; cada llamada a forward produce una fila de salida.
Parámetros dinámicos
Para consultar la referencia completa sobre formatos de firmas de funciones y tipos de datos, véase Function signatures and data types.
Los asteriscos (*) en las firmas de @annotate tienen distintos significados según su posición:
En listas de parámetros: * indica que la entrada acepta cualquier cantidad de parámetros adicionales de cualquier tipo. Por ejemplo, @annotate('double,*->string') declara un primer parámetro DOUBLE seguido de un número variable de parámetros. Debe compilar el código para calcular la cantidad y los tipos de parámetros de entrada, y gestionarlos con base en la función printf del lenguaje C. El método process recibe estos valores como *args y debe manejarlos de forma explícita.
En valores de retorno: * indica que se devuelve cualquier cantidad de valores STRING, donde el número depende de los alias proporcionados al momento de la llamada. Por ejemplo, si se llama a @annotate("bigint,string->double,*") como UDTF(x, y) as (a, b, c), se obtienen tres valores: a como DOUBLE, y b y c como STRING. La llamada a forward dentro de process debe emitir un arreglo cuya longitud coincida exactamente con la cantidad de alias; una discrepancia provoca un error en tiempo de ejecución, no en tiempo de compilación.
Esta regla del * en los valores de retorno aplica únicamente a las UDTF. Las funciones agregadas definidas por el usuario (UDAF) siempre devuelven exactamente un valor.
Código de ejemplo
Lectura de recursos desde MaxCompute
La siguiente UDTF lee asignaciones de páginas a anuncios desde un archivo de recursos JSON (test_json.txt) y una tabla de recursos (table_resource1), y luego emite una fila por cada ID de anuncio correspondiente a 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)Las funciones get_cache_file y get_cache_table (del módulo odps.distcache) cargan los recursos mediante el nombre utilizado al registrarlos con add file o add table. Los recursos se cargan una sola vez en __init__ y se reutilizan en todas las llamadas a process.
Uso de parámetros dinámicos
La siguiente UDTF analiza sintácticamente una cadena JSON y extrae valores por clave. La cantidad de valores de retorno es igual a la cantidad 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)El primer argumento (args[0]) corresponde a la cadena JSON. Los argumentos adicionales son las claves que se deben extraer. El primer valor de retorno contiene cualquier error de análisis sintáctico; los valores subsiguientes contienen el contenido extraído en el orden de las claves.
Llame a esta UDTF con un número coincidente de alias:
-- 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;Registro y llamada de una UDTF
El siguiente procedimiento utiliza UDTFExample para demostrar el flujo de trabajo completo. Guarde el código como py_udtf_example.py en la carpeta bin del cliente de MaxCompute antes de comenzar.
Paso 1: Creación de tablas de recursos y preparación de datos
Inicie sesión en el cliente de MaxCompute. Para obtener más información, consulte Start the MaxCompute client.
Cree y llene la tabla 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));El campo adid_list es de tipo ARRAY. Para permitir que Python 3 lea datos de tipo ARRAY, ejecute set odps.sql.python.version=cp37; a nivel de sesión antes de realizar consultas.
Cree y llene la tabla 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 en la carpeta bin del cliente de MaxCompute. El archivo contiene lo siguiente:
{"front_page":[1, 2, 3], "contact_page1":[3, 4, 5]}Paso 2: Registro de recursos
Agregue el archivo Python, el archivo JSON y la tabla como recursos de MaxCompute. Para obtener más información, consulte Agregar recursos.
add py py_udtf_example.py;
add file test_json.txt;
add table table_resource1 as table_resource1;Paso 3: Creación de la UDTF
Registre la UDTF y enumere todos los recursos de los que depende. Para obtener más información, consulte Crear una UDF.
create function my_udtf as 'py_udtf_example.UDTFExample' using 'py_udtf_example.py, test_json.txt, table_resource1';Paso 4: Llamada a la UDTF
Se admiten tres patrones de llamada. Todos producen la misma salida para este ejemplo.
Llamada directa:
select my_udtf(pageid) as (pageid, adid) from tmp1;Con LATERAL VIEW:
select pageid, adid from tmp1 lateral view my_udtf(pageid) adTable as udtf_pageid, adid;Con LATERAL VIEW y una función agregada:
select adid, count(1) as cnt
from tmp1 lateral view my_udtf(pageid) adTable as udtf_pageid, adid
group by adid;Las consultas de llamada directa y con LATERAL VIEW devuelven las mismas 9 filas:
+------------+------------+
| 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 |
+------------+------------+La consulta agregada devuelve lo siguiente:
+------------+------------+
| adid | cnt |
+------------+------------+
| 1 | 1 |
| 2 | 1 |
| 3 | 2 |
| 4 | 1 |
| 5 | 2 |
| 6 | 1 |
| 7 | 1 |
+------------+------------+