Scale XGBoost and Pandas Jobs to TB-level with MaxFrame - MaxCompute

MaxFrame is a distributed Python computing framework built on Alibaba Cloud MaxCompute. It provides a Pandas-compatible DataFrame API that runs across MaxCompute's elastic resource pools, letting you process terabyte (TB) to petabyte (PB) scale datasets without rewriting your existing Pandas or XGBoost code.

With MaxFrame, you can:

Run GROUP BY, JOIN, and Apply operations on PB-scale data using the same Pandas syntax you already know
Process data directly inside MaxCompute without pulling it to a local machine
Scale compute nodes on demand through MaxCompute's elastic resource scheduling
Train and run XGBoost and light gradient boosting machine (LightGBM) models on distributed data
Call large language models (LLMs) for high-concurrency offline inference through the AI Function feature

Use cases

Large-scale data analytics: Process TB/PB datasets for financial risk control, autonomous driving, or e-commerce behavior analysis. MaxFrame is dozens of times faster than single-node Pandas on this scale of data.
Interactive query and debugging: Validate data hypotheses in Jupyter Notebook using read_odps_query for SQL-like interactive access. Lazy evaluation reduces the wait time between iterations.
extract, transform, and load (ETL) pipelines: Implement log cleaning, format conversion, and other batch processing logic with apply and user-defined functions (UDFs). MaxCompute's elastic scaling adjusts resources to match job size.
Machine learning training and inference: Train XGBoost and LightGBM models on distributed data. MaxFrame provides compatible operators and handles distributed execution automatically.

When to use MaxFrame

Use MaxFrame when:

Your dataset exceeds local machine memory limits (typically beyond tens of thousands of rows)
You have existing Pandas or XGBoost code you want to scale to TB/PB data without a full rewrite
Your job requires distributed parallelism across many compute nodes

Use PyODPS instead when your dataset fits in local memory and you need a lightweight MaxCompute client without distributed overhead.

Use SQL+UDF instead when your transformation logic is expressible in SQL and you do not need Python DataFrame operations.

How it works

MaxFrame runs as a layer on top of MaxCompute. When you submit a job, MaxFrame builds a computation graph rather than executing immediately. Calling .execute() triggers the actual run, at which point MaxFrame:

Applies query optimizations such as partition pruning to reduce data scanned
Splits the job into parallel subtasks distributed across MaxCompute's compute nodes
Reads and writes data directly through MaxCompute's internal interfaces, using columnar storage and distributed cache to minimize I/O

This lazy evaluation model means you can chain multiple DataFrame operations in code and pay the I/O cost only once when you trigger execution.

Session management: Each new_session() call creates an isolated context with its own project space and compute engine version settings, so multiple jobs can run in parallel without interfering.

Prerequisites

Before you begin, make sure you have:

A MaxCompute project with the required permissions
An Alibaba Cloud AccessKey ID and AccessKey secret, set as environment variables ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET
The MaxFrame SDK installed: pip install maxframe
The PyODPS SDK installed: pip install pyodps

Quick start

The following example shows how to initialize a session, read data from a MaxCompute table, run a distributed aggregation, and fetch results.

import os
import maxframe.dataframe as md
from maxframe import new_session
from odps import ODPS

# Connect to MaxCompute
o = ODPS(
    # Store credentials in environment variables — never hardcode them.
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='<your-project>',
    endpoint='https://service.cn-<region>.maxcompute.aliyun.com/api',
)

# Create a session
session = new_session(o)

# Open this URL in a browser to monitor task progress, resource usage, and errors in real time
print(session.get_logview_address())

# Read data from a MaxCompute table via SQL
# Replace 'emp' with an existing table in your project
df = md.read_odps_query("SELECT * FROM emp")

# Run a distributed aggregation (builds a computation graph — not executed yet)
result = df.groupby("deptno").agg(total_salary=('sal', 'sum')).execute()

# Fetch the first 1,000 rows locally
local_result = result.head(1000).fetch()
print(local_result)

# Close the session
session.destroy()

Monitoring with get_logview_address(): The URL returned by this call opens the MaxCompute LogView dashboard. Use it to inspect task progress, resource usage per subtask, and error traces in real time. Open it in a browser when a job is slow or fails.

Prepare test data

If you do not already have data in your project, run the following SQL to create the sample tables used in the examples below.

-- Employee table
CREATE TABLE emp (
    empno    BIGINT COMMENT 'Employee ID',
    ename    STRING COMMENT 'Employee name',
    job      STRING COMMENT 'Job position',
    mgr      BIGINT COMMENT 'Employee ID of the manager',
    hiredate DATETIME COMMENT 'Hire date',
    sal      DECIMAL(10, 2) COMMENT 'Salary',
    comm     DECIMAL(10, 2) COMMENT 'Bonus/Commission',
    deptno   BIGINT COMMENT 'Department number'
)
COMMENT 'Employee information table'
PARTITIONED BY (
    ds STRING COMMENT 'Data partition in the YYYYMMDD format'
);

INSERT OVERWRITE TABLE emp PARTITION (ds = '20251229')
SELECT * FROM (
    VALUES
        (7369, 'SMITH', 'CLERK', 7902, CAST('1980-12-17 00:00:00' AS DATETIME), 800.00, NULL, 20),
        (7499, 'ALLEN', 'SALESMAN', 7698, CAST('1981-02-20 00:00:00' AS DATETIME), 1600.00, 300.00, 30),
        (7521, 'WARD', 'SALESMAN', 7698, CAST('1981-02-22 00:00:00' AS DATETIME), 1250.00, 500.00, 30),
        (7566, 'JONES', 'MANAGER', 7839, CAST('1981-04-02 00:00:00' AS DATETIME), 2975.00, NULL, 20),
        (7654, 'MARTIN', 'SALESMAN', 7698, CAST('1981-09-28 00:00:00' AS DATETIME), 1250.00, 1400.00, 30),
        (7698, 'BLAKE', 'MANAGER', 7839, CAST('1981-05-01 00:00:00' AS DATETIME), 2850.00, NULL, 30),
        (7782, 'CLARK', 'MANAGER', 7839, CAST('1981-06-09 00:00:00' AS DATETIME), 2450.00, NULL, 10),
        (7788, 'SCOTT', 'ANALYST', 7566, CAST('1987-04-19 00:00:00' AS DATETIME), 3000.00, NULL, 20),
        (7839, 'KING', 'PRESIDENT', NULL, CAST('1981-11-17 00:00:00' AS DATETIME), 5000.00, NULL, 10),
        (7844, 'TURNER', 'SALESMAN', 7698, CAST('1981-09-08 00:00:00' AS DATETIME), 1500.00, 0.00, 30),
        (7876, 'ADAMS', 'CLERK', 7788, CAST('1987-05-23 00:00:00' AS DATETIME), 1100.00, NULL, 20),
        (7900, 'JAMES', 'CLERK', 7698, CAST('1981-12-03 00:00:00' AS DATETIME), 950.00, NULL, 30),
        (7902, 'FORD', 'ANALYST', 7566, CAST('1981-12-03 00:00:00' AS DATETIME), 3000.00, NULL, 20),
        (7934, 'MILLER', 'CLERK', 7782, CAST('1982-01-23 00:00:00' AS DATETIME), 1300.00, NULL, 10),
        (7948, 'JACCKA', 'CLERK', 7782, CAST('1981-04-12 00:00:00' AS DATETIME), 5000.00, NULL, 10),
        (7956, 'WELAN', 'CLERK', 7649, CAST('1982-07-20 00:00:00' AS DATETIME), 2450.00, NULL, 10),
        (7957, 'TEBAGE', 'CLERK', 7748, CAST('1982-12-30 00:00:00' AS DATETIME), 1300.00, NULL, 10)
) AS t (empno, ename, job, mgr, hiredate, sal, comm, deptno);

-- Customer behavior table
CREATE TABLE IF NOT EXISTS customer_behavior (
    user_id     BIGINT COMMENT 'Unique user ID',
    action      STRING COMMENT 'User behavior: pv (page view), fav (favorite), cart (add to cart)',
    age         BIGINT COMMENT 'User age',
    gender      BIGINT COMMENT 'User gender: 0 for male, 1 for female',
    price       DOUBLE COMMENT 'Product price',
    category    STRING COMMENT 'Product category',
    buy         BIGINT COMMENT 'Whether purchased: 0 for not purchased, 1 for purchased (model label)'
)
COMMENT 'Customer behavior log table'
PARTITIONED BY (
    ds STRING COMMENT 'Data partition in the YYYYMMDD format'
);

INSERT OVERWRITE TABLE customer_behavior PARTITION (ds = '20251221')
SELECT * FROM (
    VALUES
        (1001, 'pv', 25, 1, 35.50, 'Books', 0),
        (1002, 'pv', 42, 0, 4999.00, 'Electronics', 0),
        (1003, 'pv', 31, 1, 88.00, 'Groceries', 0),
        (1004, 'pv', 19, 0, 120.00, 'Clothing', 0),
        (1005, 'fav', 28, 1, 299.00, 'Clothing', 0),
        (1006, 'fav', 35, 0, 8900.00, 'Electronics', 0),
        (1007, 'fav', 22, 1, 85.00, 'Books', 1),
        (1008, 'cart', 26, 0, 250.00, 'Sports', 0),
        (1009, 'cart', 45, 1, 1500.00, 'Home Goods', 1),
        (1010, 'cart', 33, 1, 55.00, 'Groceries', 1),
        (1011, 'cart', 29, 0, 6800.00, 'Electronics', 1),
        (1012, 'cart', 50, 0, 320.00, 'Home Goods', 1),
        (1013, 'cart', 21, 1, 450.00, 'Clothing', 0),
        (1014, 'pv', 38, 0, 99.00, 'Books', 0),
        (1015, 'pv', 23, 1, 75.00, 'Groceries', 0),
        (1016, 'cart', 30, 0, 180.00, 'Sports', 1),
        (1017, 'fav', 41, 1, 420.00, 'Home Goods', 0),
        (1018, 'pv', 65, 0, 12000.00, 'Electronics', 0),
        (1019, 'cart', 27, 1, 88.00, 'Clothing', 1),
        (1020, 'pv', 36, 0, 240.00, 'Sports', 0),
        (1021, 'cart', 39, 1, 120.00, 'Groceries', 1),
        (1022, 'fav', 24, 0, 45.00, 'Books', 0),
        (1023, 'cart', 32, 1, 3500.00, 'Electronics', 1),
        (1024, 'pv', 29, 0, 650.00, 'Home Goods', 0),
        (1025, 'cart', 48, 1, 95.00, 'Groceries', 1)
) AS t (user_id, action, age, gender, price, category, buy);

Migrate from single-node Pandas

The core change when migrating to MaxFrame is the import and session setup. Most DataFrame operations stay the same — GROUP BY, JOIN, and Apply work with the same syntax. The key difference is that data stays inside MaxCompute instead of being pulled to your local machine.

The following example shows a salary aggregation job before and after migration.

Single-node Pandas (original)

import os
import maxframe.dataframe as md
from odps import ODPS, DataFrame

o = ODPS(
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='<your-project>',
    endpoint='https://service.cn-<region>.maxcompute.aliyun.com/api',
)

# Pull data to local machine for computation
pyodps_pd = o.execute_sql("select * from emp where ds = '20251229'")
df_pd = pyodps_pd.to_pandas()                          # Data moves to your machine

result = df_pd.groupby('deptno').agg({'sal': 'sum'})

# Write result back to MaxCompute
pyodps_df = DataFrame(result)
pyodps_df.persist('result_table')

MaxFrame distributed (migrated)

import os
import maxframe.dataframe as md
from maxframe import new_session
from odps import ODPS

o = ODPS(
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='<your-project>',
    endpoint='https://service.cn-<region>.maxcompute.aliyun.com/api',
)

session = new_session(o)
print(session.get_logview_address())

# Data stays inside MaxCompute — no local transfer
df_mf = md.read_odps_query("select * from emp")        # Lazy — not executed yet

result = df_mf.groupby('deptno').agg({'sal': 'sum'})   # Still lazy

result.to_odps_table("result_table_mf")                # Triggers execution, writes to MaxCompute

session.destroy()

	Single-node Pandas	MaxFrame distributed
Import	`from odps import ODPS, DataFrame`	`from maxframe import new_session`
Read data	`o.execute_sql(...).to_pandas()` — pulls data to local machine	`md.read_odps_query(...)` — data stays in MaxCompute
Write results	`DataFrame(...).persist(...)`	`.to_odps_table(...)` — writes back distributed
Session	None required	`new_session()` / `session.destroy()`

Performance comparison

Metric	Single-node mode	MaxFrame distributed mode
Data scale	Up to tens of thousands of rows	TB/PB scale
Processing time	Hours	Minutes
Resource usage	Single-node, limited to GBs	Tens of thousands of CUs of elastic resources

Single-node XGBoost (original)

import os
import pandas as pd
import xgboost as xgb
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, roc_auc_score
from odps import ODPS

o = ODPS(
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='<your-project>',
    endpoint='https://service.cn-<region>.maxcompute.aliyun.com/api',
)

# Pull all data to local machine
pyodps_pd = o.execute_sql("select * from customer_behavior where ds = ('20251221')")
df_pd = pyodps_pd.to_pandas()

features = ["action", "age", "gender", "price", "category"]
label = "buy"

x = df_pd[features]
y = df_pd[label]
x_encoded = pd.get_dummies(x, columns=['action', 'category'], drop_first=True)

x_train, x_test, y_train, y_test = train_test_split(x_encoded, y, test_size=0.2, random_state=42)

dtrain = xgb.DMatrix(x_train, label=y_train)
dtest = xgb.DMatrix(x_test, label=y_test)

params = {
    'objective': 'binary:logistic',
    'eval_metric': 'auc',
    'learning_rate': 0.1,
    'colsample_bytree': 0.8,
    'tree_method': 'hist',
    'n_jobs': -1
}

model = xgb.train(
    params,
    dtrain,
    num_boost_round=100,
    evals=[(dtest, 'test')],
    early_stopping_rounds=10,
    verbose_eval=10
)

y_pred_proba = model.predict(dtest, iteration_range=(0, model.best_iteration))
y_pred_class = [1 if prob > 0.5 else 0 for prob in y_pred_proba]

accuracy = accuracy_score(y_test, y_pred_class)
auc = roc_auc_score(y_test, y_pred_proba)
print(f"Accuracy: {accuracy:.4f}, AUC: {auc:.4f}")

Run distributed ML training

MaxFrame includes a distributed XGBoost classifier under maxframe.learn.contrib.xgboost. The training API mirrors the standard XGBoost interface — the main difference is that input data is a MaxFrame DataFrame rather than an in-memory matrix.

All examples below use md.read_odps_table to load data and new_session to initialize MaxFrame execution.

MaxFrame distributed XGBoost (migrated)

import os
from maxframe import new_session, config
import maxframe.dataframe as md
from maxframe.learn.contrib.xgboost import XGBClassifier
from odps import ODPS

o = ODPS(
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='<your-project>',
    endpoint='https://service.cn-<region>.maxcompute.aliyun.com/api',
)

# Set the image used for UDF execution during data processing
config.options.sql.settings = {"odps.session.image": "common"}

session = new_session(o)

# Load data as a distributed DataFrame — no local transfer
all_data = md.read_odps_table("customer_behavior")

# 80% training split, full data for prediction
train_data = all_data.sample(frac=0.8, random_state=123)
predict_data = all_data

features = ["action", "age", "gender", "price", "category"]
label = "buy"

x_train = train_data[features]
y_train = train_data[label]

# MaxFrame's XGBClassifier accepts distributed DataFrames directly
model = XGBClassifier(
    n_estimators=300,
    learning_rate=0.1,
    colsample_bytree=0.8,
    n_jobs=20,          # Specify more cores — MaxCompute handles the parallelism
    tree_method="hist"
)

model.fit(x_train, y_train)                            # Builds the computation graph

predicted_result_df = model.predict(
    predict_data[features], predict_proba=True          # Returns a distributed DataFrame with predictions
)

# Merge original features with prediction results
final_df = predict_data.merge(predicted_result_df, on=predict_data.index.names)

# .head() triggers execution and fetches results to your local machine
columns_to_show = ['user_id', 'action', 'buy', 'prediction_result', 'prediction_detail']
print(final_df[columns_to_show].head(5))

session.destroy()

What changed:

Replace xgb.DMatrix + xgb.train with XGBClassifier from maxframe.learn.contrib.xgboost
Replace pd.read_csv / to_pandas() with md.read_odps_table — data stays distributed
Set n_jobs=20 — on MaxCompute you can specify a larger number of cores than a local machine provides

Implement custom logic with UDFs

When a built-in MaxFrame operator does not cover your logic, use apply or apply_chunk to run custom Python functions across distributed data.

Apply a function row by row

apply(func, axis=1) applies a function to each row. Execution is lazy — call .execute() to trigger it.

import os
import numpy as np
import maxframe.dataframe as md
from odps import ODPS
from maxframe.session import new_session

o = ODPS(
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='<your-project>',
    endpoint='https://service.cn-<region>.maxcompute.aliyun.com/api',
)

session = new_session(o)

# Create a distributed DataFrame from local data
df = md.DataFrame(np.random.randint(0, 100, size=(100, 2)), columns=["a", "b"])

def square_row(row):
    return row ** 2

result_df = df.apply(square_row, axis=1).execute()
print(result_df.fetch())

session.destroy()

Expected output:

       a     b
0   4489  7569
1   7225   841
2    441  7921
3   9604     9
4   5329   196
..   ...   ...
95    16  2209
96  6724   784
97   441  1521
98   729     1
99  3249  6889

Process data in chunks

apply_chunk(func, batch_rows=N) applies a function to fixed-size chunks of rows. Use this when your logic requires local aggregation or when operating on the full dataset at once would exceed worker memory.

import os
import numpy as np
import maxframe.dataframe as md
from odps import ODPS
from maxframe.session import new_session

o = ODPS(
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='<your-project>',
    endpoint='https://service.cn-<region>.maxcompute.aliyun.com/api',
)

session = new_session(o)

df = md.DataFrame([[4, 9]] * 3, columns=['A', 'B'])

# Sum values within each chunk of 3 rows
result = df.mf.apply_chunk(np.sum, batch_rows=3).execute()
print(result.fetch())

session.destroy()

Expected output:

A    12
B    27
dtype: int64

UDF performance notes:

Avoid importing global dependencies inside a UDF that are not installed in the MaxCompute environment. Declare them with @with_python_requirements instead.
Set batch_rows to balance memory usage against the degree of parallelism. Larger chunks use more memory per worker; smaller chunks increase scheduling overhead.

For the full apply_chunk API reference, see the apply_chunk API documentation.

Choose between MaxFrame and alternatives

Criterion	MaxFrame	PyODPS	Mars	SQL+UDF
Development interface	Pandas-compatible API	Significantly different from Pandas DataFrame	Requires separate SQL and Python interfaces	SQL only
Data processing	Data stays in MaxCompute — no local transfer	`to_pandas()` pulls data to your local machine	Distributed for some operators only; cluster startup is slow	SQL-based distributed jobs
Computing resources	Not limited by local resources; elastic scaling across MaxCompute nodes	Limited by local machine resources	Must specify worker, CPU, and memory sizes	Serverless elastic compute for SQL jobs
Development experience	Out-of-the-box interactive and offline scheduling; third-party dependencies managed via annotations	Out-of-the-box interactive and offline scheduling	Must prepare runtime environment and start a Mars cluster	Python UDF dependencies must be packaged and uploaded manually

Migration principles

When migrating existing single-node code to MaxFrame, follow these four principles:

Retain native API calls: Keep your Pandas and XGBoost method calls unchanged where MaxFrame supports them. Check the MaxFrame API reference for compatibility.
Configure distributed parameters as needed: Adjust n_jobs and other parallelism settings to take advantage of MaxCompute's compute capacity.
Keep data inside MaxCompute: Avoid intermediate steps that move data to a local machine. Use read_odps_table, read_odps_query, and to_odps_table to keep data flow within the cluster.
Use custom UDFs for unsupported operations: If a specific Pandas method is not yet supported, implement the equivalent logic with apply or apply_chunk.