MaxFrame弹性内存功能-云原生大数据计算服务 MaxCompute(MaxCompute)-阿里云帮助中心

一些场景下，数据存在偏离值时，大多数自定义函数（UDF）任务可以使用较小内存完成处理，少部分任务则需要更多内存。MaxFrame弹性内存功能即适用于处理少量数据需要较大内存的场景。开启弹性内存后，DPE 的 CPU 自定义函数（UDF）遇到系统识别的内存不足时，会自动申请更大的内存重试，最多增加到配置的上限。

适用场景

大部分数据使用较小内存即可处理，只有少量数据占用较多内存。
文件大小、文本长度或分组大小差异较大，难以提前估算峰值内存。
需要为偶发的内存不足提供自动恢复能力。

如果几乎每个任务都会出现内存不足，建议直接增大起始内存 memory，或减少每次处理的数据量，避免大部分 UDF 任务先失败再重试。

使用说明

配置弹性内存

在 UDF 的 with_running_options 中添加 memory_limit，指定重试时可使用的最大内存：

from maxframe.udf import with_running_options  

@with_running_options(engine="dpe", 
                      memory="",  # 初始内存大小，单位GiB
                      memory_limit="" # 重试可用最大内存，单位GiB
                     )
def process_batch(batch):
  #这里写数据处理逻辑
  return batch

UDF 限制

在 DPE 引擎中，下列使用 CPU 运行的 UDF 接口支持弹性内存：

处理方式	接口
对 DataFrame 的行或列执行函数	`DataFrame.apply`
对 Series 执行函数	`Series.apply`
对 DataFrame 的每个元素执行函数	`DataFrame.map`
对 DataFrame 的每个元素执行函数	`DataFrame.applymap`
分批处理 DataFrame	`DataFrame.mf.apply_chunk`
分批处理 Series	`Series.mf.apply_chunk`
分组后分批处理	`GroupBy.mf.apply_chunk`
分块处理 Tensor	`Tensor.apply_chunk`

memory 可以省略
- 若未通过 options.function.default_running_options 修改默认配置，UDF 任务从 4 GiB 开始运行；
- 若已修改默认内存，则使用配置的默认值。
memory_limit 必须大于起始内存。两者相等时，不会开启弹性内存；上限小于起始内存时，会报参数错误。
说明
实际可使用的最大内存还受平台资源限制。
例如，将起始内存设置为 4 GiB、重试时可使用的最大内存设置为 16 GiB：
```
@with_running_options(
  engine="dpe",
  memory="4GiB",         # 开始时使用的内存
  memory_limit="16GiB"   # 重试时最多使用的内存
)
def process_batch(batch):
  # 这里写数据处理逻辑
  return batch  
```

运行效果

以起始内存 4 GiB、上限 16 GiB 为例，内存不足时按以下顺序重试：

4 GiB 内存不足 → 使用 8 GiB 重试 → 仍然不足时，使用 16 GiB 重试。任一次重试成功，任务即完成；达到上限后仍然内存不足，任务会失败。实际重试还受平台资源和保护规则限制。

对于业务代码报错等其他错误，系统按普通重试规则处理，不会因此增加内存。

重要

建议保留默认的 3 次普通重试（substep.max_retries），避免偶发网络错误、磁盘错误等导致整体任务失败。

注意事项

功能仅支持上述 DPE CPU UDF 接口
UDF 使用 GU（gu_quota 或 gu_type），或在 UDF 的 options.dpe.settings 中设置 step.skip_task_resources=True 时，不会触发弹性内存功能。
仅增加目标 UDF 任务的内存
在最大内存 OOM 之前，会持续重试并放大单个 worker 申请的内存，不受 3 次重试限制。
不支持 GPU 节点。 GPU worker 独占所有内存，弹性内存功能对 GPU 节点不生效。
重试会重新执行任务。 如果函数会写文件、追加数据或调用外部服务，需要处理重复执行，避免重复写入。
达到内存上限不保证任务成功。 UDF 处理的数据过大、存在内存泄漏或平台资源不足时，任务仍可能失败或等待。增加内存和重新执行任务也可能增加耗时与资源占用。

使用示例

import os
import maxframe.dataframe as md
import numpy as np
import pandas as pd
from maxframe.config import options
from maxframe.dataframe.utils import parse_index
from maxframe.session import new_session
from maxframe.udf import with_fs_mount, with_running_options
from odps import ODPS


@with_running_options(engine="dpe", cpu=1, memory="4GiB",memory_limit="16GiB")
@with_fs_mount(
  path="oss://oss-cn-hangzhou-internal.aliyuncs.com/<bucket_name>/",
  mount_path="/mnt/maxframe_oss",
  storage_options={
    "role_arn": "acs:ram::<uid>:role/<ramrole_name>"
  },
)
def mc2oss(batch):
  import os
  import uuid

  import pandas as pd

  if batch.empty:
    return pd.DataFrame(
      {"name": pd.Series(dtype="object")},
      index=pd.Index([], dtype="int64", name="_row_id"),
    )

  relative = (
    "athangzhou_wide_100m_memray_20260911/"
    + uuid.uuid4().hex
    + ".txt"
  )
  path = "/mnt/maxframe_oss/" + relative
  os.makedirs(os.path.dirname(path), exist_ok=True)

  with open(path, "w") as stream:
    stream.write("ok")
  with open(path) as stream:
    stream.read()

  return pd.DataFrame(
    {"name": ["oss://<bucket_name>/" + relative]},
    index=pd.Index([0], name="_row_id"),
  )


options.dag.settings = {
  "engine_order": ["DPE"],
  "unavailable_engines": ["MCSQL", "SPE"],
  "python_tag": "cp311",
}
o = ODPS(
)
session = new_session(o)
input_df = md.read_odps_table(
  "mf_journal_wide_100m_20260911_at",
  columns=["journal_name", "name", "payload"],
  index_col="_row_id",
  nrows=100_000_000,
  chunk_size=100_000,
)
input_df["journal_name"] = input_df["journal_name"].fillna("---")
l = list(input_df.dtypes.index)
result_df = input_df.groupby("journal_name")[l].mf.apply_chunk(
  mc2oss,
  index=parse_index(
    pd.MultiIndex.from_product(
      [[""], [0]], names=["journal_name", "row_id"]
    )
  ),
  batch_rows=15_000,
  output_type="dataframe",
  dtypes={"name": np.str},
  prepend_index_group_keys=False,
  skip_infer=True,
)
result_df = result_df.reset_index(drop=True)
result_df.execute(session=session)

查看生效效果

在 Logview 中查看同一任务的重试记录和内存变化。未发生内存不足时，任务直接完成，不会因为开启弹性内存而额外重试。

发生 OOM 并触发弹性内存重试后，Logview 会为对应的 substep 添加内存倍率标签。例如，下图中的 [2.0x memory] 表示使用了 2 倍内存。