MaxCompute:Mount dan gunakan OSS

Aktifkan layanan dan berikan izin

1. Aktifkan OSS dan buat bucket.

   1. Masuk ke Konsol Object Storage Service (OSS).

   2. Di panel navigasi sebelah kiri, klik Buckets.

   3. Di halaman Buckets, klik Create Bucket.

      Dalam contoh ini, nama bucket adalah xxx-oss-test-sh.

2. Buat peran RAM untuk MaxCompute dan sambungkan peran tersebut ke lingkungan runtime MaxCompute.

   1. Masuk ke Konsol Resource Access Management (RAM).

   2. Di panel navigasi sebelah kiri, pilih Identities > Roles.

   3. Di halaman Roles, klik Create Role.

   4. Di pojok kanan atas halaman Create Role, klik Create Service Linked Role.

      1. Di halaman Create Role, atur Principal Type menjadi Cloud Service.

      2. Untuk Principal Name, pilih Cloud-native Big Data Computing Service MaxCompute.

      3. Di tab Permissions, klik Grant Permission. Di panel Grant Permission, pilih kebijakan akses untuk peran tersebut lalu klik OK.

         Pilih kebijakan akses berikut:

         • Izin untuk mengelola Object Storage Service (OSS): AliyunOSSFullAccess

         • Izin untuk mengelola MaxCompute: AliyunMaxComputeFullAccess

Mount OSS menggunakan with_fs_mount

1. Penggunaan yang direkomendasikan

   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. Tidak direkomendasikan

   Metode ini cocok untuk pengujian tetapi tidak untuk lingkungan produksi.

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

   Penting

   Hindari hardcoding AccessKey. Menggunakan role_arn memungkinkan sistem secara otomatis meminta token sementara Security Token Service (STS). Hal ini membantu menghindari risiko kebocoran ID AccessKey dan rahasia AccessKey Anda.

Kontrol alokasi sumber daya dengan with_running_options

Tentukan sumber daya CPU dan memori yang sesuai berdasarkan jenis tugas:

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

Buat sesi MaxFrame

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

# Inisialisasi klien ODPS
o = ODPS(
    # Pastikan variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_ID diatur ke ID AccessKey Anda.
    # Pastikan variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_SECRET diatur ke rahasia AccessKey Anda.
    # Tidak disarankan menggunakan string ID AccessKey dan rahasia AccessKey secara langsung.
    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',
)

# Atur citra runtime
# Citra maxframe_service_dpe_runtime mencakup ossfs2_2.0.3.1_linux_x86_64.deb.
# Jika Anda menggunakan custom image, unduh dependensi OSS, lalu unggah dan gunakan dalam citra tersebut. Paket dependensi tercantum di bawah blok kode ini.
options.sql.settings = { "odps.session.image": "maxframe_service_dpe_runtime"}

# Mulai sesi
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"
    },
)

Paket dependensi ossfs: ossfs2_2.0.3.1_linux_x86_64.deb

Buat fungsi yang ditentukan pengguna

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

  # Langkah 1: Periksa apakah pemasangan berhasil
  mount_point = "/mnt/oss_data"
  if not os.path.exists(mount_point):
    raise RuntimeError("OSS mount failed!")

    # Langkah 2: Muat data (seperti tabel pemetaan atau kamus)
  mapping_file = os.path.join(mount_point, "category_map.csv")
  if os.path.isfile(mapping_file):
    mapping_df = pd.read_csv(mapping_file)

    # Langkah 3: Proses chunk saat ini
  result = batch_df.copy()
  result['F'] = result['A'] * 10

  return result

Buat DataFrame dan terapkan fungsi yang ditentukan pengguna

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

# Gunakan apply_chunk untuk menerapkan fungsi setelah pemasangan
result_df = df.mf.apply_chunk(
  _process,
  skip_infer=True,
  output_type="dataframe",
  dtypes=df.dtypes,
  index=df.index
)

# Eksekusi dan ambil hasilnya
result = result_df.execute().fetch()

skip_infer=True melewatkan inferensi tipe, yang mempercepat eksekusi. Pastikan Anda meneruskan dtypes dan index dengan benar.

Verifikasi status pemasangan

Tambahkan log debugging ke fungsi _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 [])

Periksa output LogView untuk log yang mirip dengan berikut:

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