MaxCompute: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
   import os
   
   o = ODPS(
       # Ensure that the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is set to your AccessKey ID.
       # Ensure that the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is set to your AccessKey secret.
       # We recommend that you do not directly use the strings of your AccessKey ID and 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シークレットに設定します。

   • your-default-project: MaxComputeプロジェクトの名前。 プロジェクト名を表示するには、MaxComputeコンソールにログインし、左側のナビゲーションウィンドウから [ワークスペース] > [プロジェクト] を選択します。

   • your-end-point: MaxComputeプロジェクトのリージョンのエンドポイントです。 http://service.cn-chengdu.maxcompute.aliyun.com/api などのネットワーク接続方法に基づいて選択できます。 詳細については、「エンドポイント」をご参照ください。

2. 次のSQLコマンドを使用して、sales_maxframe_demoおよびproduct_maxframe_demoテーブルからデータを取得します。

   --Query sales_maxframe_demo table
   SELECT * FROM sales_maxframe_demo;
   --Return results
   +------------+------------+------------+------------+------------+------------+------------+
   | 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 product_maxframe_demo table data
   SELECT * FROM product_maxframe_demo;
   --Return results
   +------------+------------+--------------+---------------+
   | index      | product_id | product_name | current_price |
   +------------+------------+--------------+---------------+
   | 1          | 100        | Nokia        | 1000          |
   | 2          | 200        | Apple        | 5000          |
   | 3          | 300        | Samsung      | 9000          |
   +------------+------------+--------------+---------------+

データ分析にMaxFrameを使用する

結論

MaxFrameはPandasインターフェイスとシームレスに統合され、自動分散処理を可能にします。 堅牢なデータ処理機能を維持しながら、データ分析と計算の規模と効率を大幅に向上させます。