MaxCompute:Mejores prácticas para usar el operador apply_chunk de MaxFrame

DataFrame.mf.apply_chunk(
    func,
    batch_rows=None,
    output_type=None,
    dtypes=None,
    index=None,
    index_value=None,
    columns=None,
    elementwise=None,
    sort=False,
    **kwds
)

Descripción de dtypes

dtypes es un objeto pd.Series que describe los nombres de columna y los tipos de datos de la salida de su UDF. MaxFrame lo utiliza para validar y combinar resultados entre workers. Si dtypes no coincide con lo que realmente devuelve func, el trabajo falla durante la ejecución.

Para inspeccionar la estructura de dtypes en un DataFrame:

>>> df.dtypes
A    object
B    object
dtype: object

La forma más común de proporcionar dtypes es copiarlo del DataFrame original:

dtypes=df.dtypes.copy()

Si su UDF modifica el esquema de salida (agrega o elimina columnas), construya un nuevo objeto pd.Series que describa la salida real:

import pandas as pd

# UDF drops column 'A' and adds column 'C'
new_dtypes = pd.Series({
    "B": df.dtypes["B"],
    "C": pd.StringDtype(),
})

import os
import pyarrow as pa
import pandas as pd
import maxframe.dataframe as md
from maxframe.lib.dtypes_extension import dict_
from maxframe import new_session
from odps import ODPS

o = ODPS(
    # Get credentials from environment variables.
    # Never hardcode AccessKey ID or AccessKey secret in your code.
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='<your-project>',
    endpoint='https://service.cn-<your-region>.maxcompute.aliyun.com/api',
)

session = new_session(o)

# Create a MaxFrame DataFrame with a dict-type column
col_a = pd.Series(
    data=[[("k1", 1), ("k2", 2)], [("k1", 3)], None],
    index=[1, 2, 3],
    dtype=dict_(pa.string(), pa.int64()),
)
col_b = pd.Series(
    data=["A", "B", "C"],
    index=[1, 2, 3],
)
df = md.DataFrame({"A": col_a, "B": col_b})
df.execute()

def custom_set_item(df: pd.DataFrame) -> pd.DataFrame:
    """Add key 'x' with value 100 to each non-null dict in column A."""
    for name, value in df["A"].items():
        if value is not None:
            df["A"][name]["x"] = 100
    return df

result_df = df.mf.apply_chunk(
    custom_set_item,
    output_type="dataframe",
    dtypes=df.dtypes.copy(),   # Must match the columns returned by the UDF
    batch_rows=2,              # Process 2 rows per chunk
    skip_infer=True,
    index=df.index,
).execute()

session.destroy()

Establezca batch_rows según sus datos y recursos

batch_rows controla cuántas filas procesa cada worker por fragmento. El valor adecuado depende del tamaño de sus filas y de la memoria disponible:

Comience con un valor conservador, supervise el uso de memoria en LogView y ajústelo según sea necesario. Si observa errores OOM, reduzca batch_rows. Si los trabajos se ejecutan rápidamente con bajo uso de memoria, auméntelo.

Si su UDF requiere más memoria por fragmento, use el decorador @with_running_options para aumentar el límite de memoria por tarea:

from maxframe.udf import with_running_options

@with_running_options(memory=16)
def my_udf(df: pd.DataFrame) -> pd.DataFrame:
    ...

Declare siempre output_type y dtypes explícitamente

De forma predeterminada, MaxFrame intenta inferir el esquema de salida ejecutando su función sobre datos de muestra. La inferencia puede fallar o producir resultados incorrectos en esquemas complejos. Declarar explícitamente output_type y dtypes evita fallos en tiempo de ejecución y acelera el trabajo.

Sin declaración explícita (falla en tiempo de ejecución):

result_df = df.mf.apply_chunk(process)  # dtypes is missing

Con declaración explícita (correcto):

result_df = df.mf.apply_chunk(
    process,
    output_type="dataframe",
    dtypes=df.dtypes.copy(),
)

Un error común es pasar dtypes que no coinciden con la salida real de la UDF. Por ejemplo, si elimina una columna dentro de func pero sigue pasando el df.dtypes original, MaxFrame genera un ValueError porque el esquema declarado tiene más columnas que el DataFrame devuelto.

Devuelva solo las columnas necesarias

Cada columna de salida se serializa, transfiere y combina entre workers. Devolver columnas no utilizadas aumenta el uso de memoria y ralentiza el trabajo. Dentro de func, elimine cualquier columna que no necesite antes de devolver el resultado.

Depure UDF con print y flush=True

Las UDF se ejecutan en workers remotos de MaxCompute. Use print(..., flush=True) para emitir registros de inmediato; estos aparecerán en LogView. Envuelva el cuerpo de la UDF en un bloque try/except para capturar y registrar errores antes de que se manifiesten como fallos genéricos:

def process(chunk: pd.DataFrame) -> pd.DataFrame:
    try:
        print(f"Processing chunk: shape={chunk.shape}, columns={list(chunk.columns)}", flush=True)
        result = chunk.sort_values("B")
        print("Chunk processed successfully.", flush=True)
        return result
    except Exception as e:
        print(f"[ERROR] {type(e).__name__}: {e}", flush=True)
        raise

¿Por qué obtengo TypeError: cannot determine dtype?

MaxFrame no pudo inferir el esquema de salida. Pase dtypes y output_type explícitamente al llamar a apply_chunk.

¿Por qué mi salida está vacía o le faltan columnas?

Los dtypes que pasó no coinciden con lo que devuelve su UDF. Compruebe que los nombres de columna en dtypes coincidan exactamente con las columnas del DataFrame que devuelve func. Elimine de dtypes cualquier columna que func no incluya en su valor de retorno.

¿Por qué el trabajo se cuelga o supera el tiempo de espera?

Probablemente batch_rows sea demasiado grande. Redúzcalo y asigne más recursos para que cada worker procese un fragmento más pequeño y rápido. Revise también LogView en busca de errores OOM, ya que a menudo se manifiestan como cuelgues en lugar de fallos explícitos.