Mounting and using OSS

Activate services and grant permissions

1. Activate OSS and create a bucket.

   1. Log in to the OSS console.

   2. In the left-side navigation pane, click Buckets.

   3. On the Buckets page, click Create Bucket.

      In this example, the bucket name is xxx-oss-test-sh.

2. Create a RAM role for MaxCompute and grant it access to the runtime environment.

   1. Log in to the RAM console.

   2. In the left navigation bar, select Identities > Roles.

   3. On the Roles page, click Create Role.

   4. In the upper-right corner of the Create Role page, click Create Service Linked Role.

      1. On the Create Role page, set Principal Type to Cloud Service.

      2. For Principal Type, select MaxCompute.

      3. On the Manage Permissions tab, click Create Authorization. In the Create Authorization panel that appears, select the policies to grant to the role, and click OK.

         Select the following policies:

         • AliyunOSSFullAccess: Grants permissions to manage OSS.

         • AliyunMaxComputeFullAccess: Grants permissions to manage MaxCompute.

Use with_fs_mount to mount OSS

1. Recommended: Authenticate with a role ARN

   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
       

2. Not recommended: Hard-code credentials

   This method is for testing only and not recommended for production environments.

   storage_options={
       "access_key_id": "LTAI5t...",
       "access_key_secret": "Wp9H..."
   }

   Important

   Avoid hard-coding your AccessKey. You can use role_arn to allow the system to automatically request a temporary STS token, preventing your AccessKey pair from being leaked.

Use with_running_options to control resource allocation

Use the with_running_options decorator to allocate appropriate CPU and memory resources for your task type:

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

Create a MaxFrame session

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

# Initialize the ODPS client.
# We recommend setting the ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET
# environment variables instead of hard-coding the AccessKey ID and AccessKey Secret strings.
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 runtime image.
# The `maxframe_service_dpe_runtime` image includes the required ossfs2 dependency.
# If you use a custom image, you must download the dependency and include it in your image.
# You can find the package at the link below this code block.
options.sql.settings = { "odps.session.image": "maxframe_service_dpe_runtime"}

# Start the session.
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 dependency package: ossfs2_2.0.3.1_linux_x86_64.deb

Create a user-defined function (UDF)

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

  # Step 1: Check if the mount was successful.
  mount_point = "/mnt/oss_data"
  if not os.path.exists(mount_point):
    raise RuntimeError("OSS mount failed!")

    # Step 2: Load data, such as a mapping table or dictionary.
  mapping_file = os.path.join(mount_point, "category_map.csv")
  if os.path.isfile(mapping_file):
    mapping_df = pd.read_csv(mapping_file)

    # Step 3: Process the current chunk.
  result = batch_df.copy()
  result['F'] = result['A'] * 10

  return result

Build a DataFrame and apply the UDF

import maxframe.dataframe as md

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

# Apply the UDF by using apply_chunk.
result_df = df.mf.apply_chunk(
  _process,
  skip_infer=True,
  output_type="dataframe",
  dtypes=df.dtypes,
  index=df.index
)

# Execute the operation and fetch the result.
result = result_df.execute().fetch()

You can set skip_infer=True to skip type inference and improve execution speed. However, you must ensure that dtypes and index are passed correctly.

Verify the mount status

You can add debugging logs in the _process function:

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 [])

Check the LogView output to confirm that logs similar to the following are generated:

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