Distributed pandas processing with MaxFrame-MaxCompute(MaxCompute)-阿里云帮助中心

Prerequisites

MaxFrame is installed. For more information, see Preparations.

Data preparation

Run the following script in a Python environment where MaxFrame is installed to prepare the test tables and data.

from odps import ODPS
from maxframe.session import new_session
import maxframe.dataframe as md
import pandas as pd
import os

o = ODPS(
    # Make sure the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is set to your AccessKey ID,
    # and the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is set to your AccessKey Secret.
    # We do not recommend using AccessKey ID and AccessKey Secret strings directly in your code.
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='your-default-project',
    endpoint='your-end-point',
)

data_sets = [{
    "table_name": "product",
    "table_schema" : "index bigint, product_id bigint, product_name string, current_price bigint",
    "source_type": "records",
    "records" : [
        [1, 100, 'Nokia', 1000],
        [2, 200, 'Apple', 5000],
        [3, 300, 'Samsung', 9000]
    ],
},
{
    "table_name" : "sales",
    "table_schema" : "index bigint, sale_id bigint, product_id bigint, user_id bigint, year bigint, quantity bigint, price bigint",
    "source_type": "records",
    "records" : [
        [1, 1, 100, 101, 2008, 10, 5000],
        [2, 2, 300, 101, 2009, 7, 4000],
        [3, 4, 100, 102, 2011, 9, 4000],
        [4, 5, 200, 102, 2013, 6, 6000],
        [5, 8, 300, 102, 2015, 10, 9000],
        [6, 9, 100, 102, 2015, 6, 2000]
    ],
    "lifecycle": 5
}]

def prepare_data(o: ODPS, data_sets, suffix="", drop_if_exists=False):
    for index, data in enumerate(data_sets):
        table_name = data.get("table_name")
        table_schema = data.get("table_schema")
        source_type = data.get("source_type")
        
        if not table_name or not table_schema or not source_type:
            raise ValueError(f"Dataset at index {index} is missing one or more required keys: 'table_name', 'table_schema', or 'source_type'.")

        lifecycle = data.get("lifecycle", 5)
        table_name += suffix
        
        print(f"Processing {table_name}...")
        if drop_if_exists:
            print(f"Deleting {table_name}...")
            o.delete_table(table_name, if_exists=True)
        
        o.create_table(name=table_name, table_schema=table_schema, lifecycle=lifecycle, if_not_exists=True)

        if source_type == "local_file":
            file_path = data.get("file")
            if not file_path:
                raise ValueError(f"Dataset at index {index} with source_type 'local_file' is missing the 'file' key.")
            sep = data.get("sep", ",")
            pd_df = pd.read_csv(file_path, sep=sep)
            ODPSDataFrame(pd_df).persist(table_name, drop_table=True)
        elif source_type == 'records':
            records = data.get("records")
            if not records:
                raise ValueError(f"Dataset at index {index} with source_type 'records' is missing the 'records' key.")
            with o.get_table(table_name).open_writer() as writer:
                writer.write(records)
        else:
            raise ValueError(f"Unknown data set source_type: {source_type}")
        
        print(f"Processed {table_name} Done")

prepare_data(o, data_sets, "_maxframe_demo", True)

Parameters:

ALIBABA_CLOUD_ACCESS_KEY_ID: Set this environment variable to an AccessKey ID with the required MaxCompute permissions for the objects in your target MaxCompute project. You can obtain an AccessKey ID from the AccessKey Management page.
ALIBABA_CLOUD_ACCESS_KEY_SECRET: Set this environment variable to the AccessKey Secret that corresponds to the AccessKey ID.
your-default-project: The name of your MaxCompute project. You can find it on the MaxCompute console under Workspace > Projects in the left-side navigation pane.
your-end-point: The endpoint of the region where your MaxCompute project is located. Select an endpoint based on your network connection type, for example, http://service.cn-chengdu.maxcompute.aliyun.com/api. For more information, see Endpoints.

Query the data in the sales_maxframe_demo and product_maxframe_demo tables using the following SQL commands.

--Query the sales_maxframe_demo table
SELECT * FROM sales_maxframe_demo;

--Output
+------------+------------+------------+------------+------------+------------+------------+
| index      | sale_id    | product_id | user_id    | year       | quantity   | price      |
+------------+------------+------------+------------+------------+------------+------------+
| 1          | 1          | 100        | 101        | 2008       | 10         | 5000       |
| 2          | 2          | 300        | 101        | 2009       | 7          | 4000       |
| 3          | 4          | 100        | 102        | 2011       | 9          | 4000       |
| 4          | 5          | 200        | 102        | 2013       | 6          | 6000       |
| 5          | 8          | 300        | 102        | 2015       | 10         | 9000       |
| 6          | 9          | 100        | 102        | 2015       | 6          | 2000       |
+------------+------------+------------+------------+------------+------------+------------+


--Query the product_maxframe_demo table data
SELECT * FROM product_maxframe_demo;

--Output
+------------+------------+--------------+---------------+
| index      | product_id | product_name | current_price |
+------------+------------+--------------+---------------+
| 1          | 100        | Nokia        | 1000          |
| 2          | 200        | Apple        | 5000          |
| 3          | 300        | Samsung      | 9000          |
+------------+------------+--------------+---------------+

Analyze data using MaxFrame

Scenario 1: Use the merge method to join two data tables to retrieve from the sales_maxframe_demo table all sale_ids and their corresponding product_names, as well as all associated year and price.

Sample code

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

o = ODPS(
    # Make sure the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is set to your AccessKey ID,
    # and the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is set to your AccessKey Secret.
    # We do not recommend using AccessKey ID and AccessKey Secret strings directly in your code.
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='your-default-project',
    endpoint='your-end-point',
)

session = new_session(o)

# The session ID is a string that associates MaxFrame tasks and is crucial for debugging and tracking their status.
print(session.session_id)

sales = md.read_odps_table("sales_maxframe_demo", index_col="index")
product = md.read_odps_table("product_maxframe_demo", index_col="product_id")

# Because operations are lazily evaluated, the df DataFrame is not executed immediately.
# You must call df.execute() to trigger the computation.
# All computations are performed on the MaxCompute cluster, avoiding unnecessary data transfers and blocking.
df = sales.merge(product, left_on="product_id", right_index=True)
df = df[["product_name", "year", "price"]]

print(df.execute().fetch())

# Save the results to a MaxCompute table and then destroy the session.
md.to_odps_table(df, "result_df", overwrite=True).execute()

session.destroy()

Output:

index product_name  year  price                   
1            Nokia  2008   5000
2          Samsung  2009   4000
3            Nokia  2011   4000
4            Apple  2013   6000
5          Samsung  2015   9000
6            Nokia  2015   2000

Performance comparison

For a sample dataset with 50 million records in the sales table (1.96 GB) and 100,000 records in the product table (3 MB), the performance is compared below:

Environment	Execution time
Local Pandas (v1.3.5)	65.8
MaxFrame	22

Scenario 2: Find the sales data for the first year each product was sold

Sample code

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

o = ODPS(
    # Make sure the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is set to your AccessKey ID,
    # and the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is set to your AccessKey Secret.
    # We do not recommend using AccessKey ID and AccessKey Secret strings directly in your code.
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='your-default-project',
    endpoint='your-end-point',
)

session = new_session(o)

# The session ID is a string that associates MaxFrame tasks and is crucial for debugging and tracking their status.
print(session.session_id)

# Aggregate to get the first year for each product.
min_year_df = md.read_odps_table("sales_maxframe_demo", index_col="index")
min_year_df = min_year_df.groupby('product_id', as_index=False).agg(first_year=('year', 'min'))

# Join to find the corresponding sales records.
sales = md.read_odps_table("sales_maxframe_demo", index_col=['product_id', 'year'])
result_df = md.merge(sales, min_year_df, 
                        left_index=True, 
                        right_on=['product_id','first_year'],
                        how='inner')

# Because operations are lazily evaluated, the result_df DataFrame is not executed immediately.
# You must call result_df.execute() to trigger the computation.
# All computations are performed on the MaxCompute cluster, avoiding unnecessary data transfers and blocking.
result_df = result_df[['product_id', 'first_year', 'quantity', 'price']]

print(result_df.execute().fetch())

# Destroy the session.
session.destroy()

Output:

    product_id  first_year  quantity  price
100         100        2008        10   5000
300         300        2009         7   4000
200         200        2013         6   6000

Performance comparison

For a sample dataset with 50 million records in the sales table (1.96 GB) and 100,000 records in the product table (3 MB), the performance is compared below:

Environment	Execution time
Local Pandas (v1.3.5)	186
MaxFrame	21

Scenario 3: Find the product on which each user spent the most

Note

This scenario demonstrates a series of operations, including groupby, join, drop_duplicates, and sort_values.

Sample code

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

o = ODPS(
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'),
    os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'),
    project='your-default-project',
    endpoint='your-end-point',
)

session = new_session(o)

# The session ID is a string that associates MaxFrame tasks and is crucial for debugging and tracking their status.
print(session.session_id)

sales = md.read_odps_table("sales_maxframe_demo", index_col="index")
product = md.read_odps_table("product_maxframe_demo", index_col="product_id")

sales['total'] = sales['price'] * sales['quantity']

product_cost_df = sales.groupby(['product_id', 'user_id'], as_index=False).agg(user_product_total=('total','sum'))
product_cost_df = product_cost_df.merge(product, left_on="product_id", right_index=True, how='right')

user_cost_df = product_cost_df.groupby('user_id').agg(max_total=('user_product_total', 'max'))
merge_df = product_cost_df.merge(user_cost_df, left_on='user_id', right_index=True)

# Because operations are lazily evaluated, the result_df DataFrame is not executed immediately.
# You must call result_df.execute() to trigger the computation.
# All computations are performed on the MaxCompute cluster, avoiding unnecessary data transfers and blocking.
result_df = merge_df[merge_df['user_product_total'] == merge_df['max_total']][['user_id', 'product_id']].drop_duplicates().sort_values(['user_id'], ascending = [1])

print(result_df.execute().fetch())

# Destroy the session.

session.destroy()

Output:

     user_id  product_id
100      101         100
300      102         300

Performance comparison

For a sample dataset with 50 million records in the sales table (1.96 GB) and 100,000 records in the product table (3 MB), the performance is compared below:

Environment	Execution time
Local Pandas (v1.3.5)	176
MaxFrame	85

Conclusion

MaxFrame is compatible with the Pandas API and automatically handles distributed processing. This provides robust data processing and significantly increases the scale and efficiency of your computations.