効率的な MaxFrame 分散データアクセス用 OSS マウント - MaxCompute - Alibaba Cloud - MaxCompute

このトピックでは、実践的なコード例を使用して、MaxFrame での分散コンピューティングのストレージとして Alibaba Cloud Object Storage Service (OSS) を効率的かつ安全にマウントして使用する方法を説明します。 with_fs_mount デコレーターは、ファイルシステムレベルのマウントを有効にすることで、大規模なデータ処理向けに安定した信頼性の高い外部データアクセスを提供します。

利用シーン

この方法は、MaxFrame のタスクと OSS のような永続的なオブジェクトストレージを組み合わせるビッグデータ分析シナリオに適用できます。例：

OSS から生データをロードして、クリーニングや処理を行う。
中間結果を OSS に書き込み、後続タスクで利用する。
トレーニング済みモデルファイルや構成ファイルなどの静的リソースを共有する。

pd.read_csv("oss://...") のような従来の読み書きメソッドは、SDK のパフォーマンスとネットワークのオーバーヘッドによって制限されるため、分散環境では非効率です。ファイルシステムレベルのマウント (FS Mount) を使用すると、MaxCompute 内の OSS ファイルに、ローカルディスク上にあるかのようにアクセスできます。これにより、開発効率が大幅に向上します。

ガイド

サービスの有効化と権限の付与

OSS を有効化し、バケットを作成します。
1. Object Storage Service (OSS) コンソールにログインします。
2. 左側のナビゲーションウィンドウで、バケット をクリックします。
3. バケット ページで、バケットの作成 をクリックします。
  この例では、バケット名は xxx-oss-test-sh です。
MaxCompute 用の RAM ロールを作成し、そのロールを MaxCompute の実行環境にアタッチします。
1. Resource Access Management (RAM) コンソールにログインします。
2. 左側のナビゲーションウィンドウで、[アイデンティティ] > [ロール]を選択します。
3. [ロール] ページで、[ロールの作成] をクリックします。
4. [ロールの作成] ページの右上隅で、[サービスリンクロールの作成] をクリックします。
  1. [ロールの作成] ページで、[信頼できるエンティティの種類] を [クラウドサービス] に設定します。
  2. [プリンシパル名] には、[クラウドネイティブビッグデータコンピューティングサービス MaxCompute] を選択します。
  3. [権限] タブで、[権限の付与] をクリックします。[権限の付与] パネルで、ロールのアクセスポリシーを選択し、[OK] をクリックします。
    以下のアクセスポリシーを選択します：
    - Object Storage Service (OSS) を管理する権限： AliyunOSSFullAccess
    - MaxCompute を管理する権限： AliyunMaxComputeFullAccess

`with_fs_mount` を使用した OSS のマウント

推奨される使用方法

from maxframe.udf import with_fs_mount

@with_fs_mount(
    "oss://oss-cn-xxxx-internal.aliyuncs.com/xxx-oss-test-sh/test/",
    "/mnt/oss_data",
    storage_options={
        "role_arn": "acs:ram::xxx:role/maxframe-oss"
    },
)
def _process(batch_df):
    import os
    if os.path.exists('/mnt/oss_data'):
        print(f"Mounted files: {os.listdir('/mnt/oss_data')}")
    else:
        print("/mnt/oss_data not mounted!")
    return batch_df * 2

非推奨
この方法はテストには適していますが、本番環境には適していません。
```
storage_options={
    "access_key_id": "LTAI5t...",
    "access_key_secret": "Wp9H..."
}
```
重要
AccessKey のハードコーディングは避けてください。role_arn を使用すると、システムが一時的な STS トークンを自動的にリクエストできます。これにより、AccessKey ID と AccessKey Secret が漏洩するリスクを回避することができます。

`with_running_options` によるリソース割り当ての制御

タスクの種類に応じて、適切な CPU とメモリリソースを設定します：

from maxframe.udf import with_running_options
@with_running_options(engine="dpe", cpu=2, memory=16)
@with_fs_mount(...)
def _process(batch_df):
    ...

パラメーター	推奨値	説明
`engine="dpe"`	固定	現在、FS Mount は DPE エンジンのみをサポートしています。
`cpu`	1～4	複雑な I/O 操作や展開を行う場合は、この値を大きくします。
`memory`	8 GB 以上	大きなファイルをロードする場合は、16 GB 以上を推奨します。

使用例

推奨パターン： データバッチ処理。

大規模なデータ処理では、MaxFrame の apply_chunk 機能を使用して、入力データをバッチで処理します。

MaxFrame セッションの作成

import os
from odps import ODPS
from maxframe import new_session
from maxframe.udf import with_fs_mount

# ODPS クライアントを初期化します
o = ODPS(
    # 環境変数 ALIBABA_CLOUD_ACCESS_KEY_ID がご自身の AccessKey ID に設定されていることを確認してください。
    # 環境変数 ALIBABA_CLOUD_ACCESS_KEY_SECRET がご自身の AccessKey Secret に設定されていることを確認してください。
    # AccessKey ID と AccessKey Secret の文字列を直接使用することは推奨されません。
    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',
)

# 実行イメージを設定します
# maxframe_service_dpe_runtime イメージには ossfs2_2.0.3.1_linux_x86_64.deb が含まれています。
# カスタムイメージを使用する場合は、OSS の依存関係をダウンロードし、アップロードしてイメージ内で使用してください。依存関係パッケージはこのコードブロックの下にリストされています。
options.sql.settings = { "odps.session.image": "maxframe_service_dpe_runtime"}

# セッションを開始します
session = new_session(o)

print("LogView:", session.get_logview_address())
print("Session ID:", session.session_id)

@with_running_options(engine="dpe", cpu=2, memory=8)
@with_fs_mount(
    "oss://oss-cn-<region>-internal.aliyuncs.com/wzy-oss-test-sh/test/",
    "/mnt/oss_data",
    storage_options={
        "role_arn": "acs:ram::<uid>:role/maxframe-oss"
    },
)

ossfs 依存関係パッケージ： ossfs2_2.0.3.1_linux_x86_64.deb

ユーザー定義関数の作成

def _process(batch_df):
  import pandas as pd
  import os

  # ステップ 1： マウントが成功したか確認します
  mount_point = "/mnt/oss_data"
  if not os.path.exists(mount_point):
    raise RuntimeError("OSS mount failed!")

    # ステップ 2： データをロードします (マッピングテーブルや辞書など)
  mapping_file = os.path.join(mount_point, "category_map.csv")
  if os.path.isfile(mapping_file):
    mapping_df = pd.read_csv(mapping_file)

    # ステップ 3： 現在のチャンクを処理します
  result = batch_df.copy()
  result['F'] = result['A'] * 10

  return result

DataFrame の構築とユーザー定義関数の適用

data = [[1.0, 2.0, 3.0, 4.0, 5.0], ...]
df = md.DataFrame(data, columns=['A', 'B', 'C', 'D', 'E'])

# マウント後、apply_chunk を使用して関数を適用します
result_df = df.mf.apply_chunk(
  _process,
  skip_infer=True,
  output_type="dataframe",
  dtypes=df.dtypes,
  index=df.index
)

# 実行して結果を取得します
result = result_df.execute().fetch()

skip_infer=True を指定すると、型推論がスキップされ、実行が高速化されます。dtypes と index を正しく渡すようにしてください。

デバッグのヒント

マウント状態の確認

_process 関数にデバッグログを追加します：

import os
print("Mount path exists:", os.path.exists("/mnt/oss_data"))
print("Files in mount:", os.listdir("/mnt/oss_data") if os.path.exists("/mnt/oss_data") else [])

LogView の出力で、以下のようなログを確認します：

FS Mount successful! /mnt/oss_data: ['data.csv', 'config.json', 'model.pkl']
Processing batch with shape: (1000, 5)