基于MaxFrame实现分布式Pandas处理

前提条件

已安装MaxFrame，详情请参见准备工作。

数据准备

1. 在安装了MaxFrame的Python环境下运行如下脚本，准备测试表和测试数据。

   from odps import ODPS
   from maxframe.session import new_session
   import maxframe.dataframe as md
   import pandas as pd
   
   o = ODPS(
       # 确保 ALIBABA_CLOUD_ACCESS_KEY_ID 环境变量设置为用户 Access Key ID，
       # ALIBABA_CLOUD_ACCESS_KEY_SECRET 环境变量设置为用户 Access Key Secret，
       # 不建议直接使用AccessKey ID和 AccessKey Secret字符串。
       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)

   参数说明：

   • ALIBABA_CLOUD_ACCESS_KEY_ID：需将该环境变量设置为具备目标MaxCompute项目中待操作对象相关MaxCompute权限的AccessKey ID。您可以进入AccessKey管理页面获取AccessKey ID。

   • ALIBABA_CLOUD_ACCESS_KEY_SECRET：需将该环境变量设置为AccessKey ID对应的AccessKey Secret。

   • your-default-project：使用的MaxCompute项目名称。您可以登录MaxCompute控制台，在左侧导航栏选择工作区>项目管理，查看MaxCompute项目名称。

   • your-end-point：目标MaxCompute项目所在地域的Endpoint，可根据网络连接方式自行选择，例如http://service.cn-chengdu.maxcompute.aliyun.com/api。详情请参见Endpoint。

2. 查询sales_maxframe_demo表和product_maxframe_demo表的数据，SQL命令如下。

   --查询sales_maxframe_demo表
   SELECT * FROM sales_maxframe_demo;
   
   --返回
   +------------+------------+------------+------------+------------+------------+------------+
   | 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       |
   +------------+------------+------------+------------+------------+------------+------------+
   
   
   --查询product_maxframe_demo表数据
   SELECT * FROM product_maxframe_demo;
   
   --返回
   +------------+------------+--------------+---------------+
   | index      | product_id | product_name | current_price |
   +------------+------------+--------------+---------------+
   | 1          | 100        | Nokia        | 1000          |
   | 2          | 200        | Apple        | 5000          |
   | 3          | 300        | Samsung      | 9000          |
   +------------+------------+--------------+---------------+

使用MaxFrame进行数据分析

结论

MaxFrame兼容Pandas接口且自动进行分布式处理，在保证强大数据处理能力的同时，可以大幅度提高数据处理规模及计算效率。