MaxCompute:Memasang dan menggunakan OSS

Aktifkan layanan dan berikan izin

1. Aktifkan OSS dan buat bucket.

   1. Login ke OSS console.

   2. Di panel navigasi sebelah kiri, klik Buckets.

   3. Di halaman Buckets, klik Create Bucket.

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

2. Buat peran RAM untuk MaxCompute dan berikan akses ke lingkungan runtime.

   1. Login ke RAM console.

   2. Di bilah navigasi 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 Type, pilih MaxCompute.

      3. Di tab Manage Permissions, klik Create Authorization. Di panel Create Authorization yang muncul, pilih kebijakan yang akan diberikan kepada peran tersebut, lalu klik OK.

         Pilih kebijakan berikut:

         • AliyunOSSFullAccess: Memberikan izin untuk mengelola OSS.

         • AliyunMaxComputeFullAccess: Memberikan izin untuk mengelola MaxCompute.

Gunakan with_fs_mount untuk memasang OSS

1. Disarankan: Lakukan autentikasi dengan 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. Tidak disarankan: Hard-code kredensial

   Metode ini hanya untuk pengujian dan tidak disarankan untuk lingkungan produksi.

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

   Penting

   Hindari hard-coding AccessKey Anda. Anda dapat menggunakan role_arn agar sistem secara otomatis meminta token STS temporary, mencegah Pasangan Kunci Akses Anda bocor.

Gunakan with_running_options untuk mengontrol alokasi sumber daya

Gunakan dekorator with_running_options untuk mengalokasikan sumber daya CPU dan memori bagi tugas Anda:

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, with_running_options

# Inisialisasi klien ODPS.
# Kami menyarankan menyetel variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_ID dan ALIBABA_CLOUD_ACCESS_KEY_SECRET
# daripada hard-coding string ID AccessKey dan AccessKey Secret.
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',
)

# Setel citra runtime.
# Citra `maxframe_service_dpe_runtime` mencakup dependensi ossfs2 yang diperlukan.
# Jika Anda menggunakan gambar kustom, Anda harus mengunduh dependensi tersebut dan menyertakannya dalam citra Anda.
# Anda dapat menemukan paket tersebut di tautan 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 user-defined (UDF)

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

# Terapkan UDF dengan menggunakan apply_chunk.
result_df = df.mf.apply_chunk(
  _process,
  skip_infer=True,
  output_type="dataframe",
  dtypes=df.dtypes,
  index=df.index
)

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

Menyetel skip_infer=True melewatkan inferensi tipe dan meningkatkan kecepatan eksekusi. Namun, Anda harus memastikan bahwa dtypes dan index diteruskan dengan benar.

Verifikasi status pemasangan

Tambahkan log debugging di 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 memastikan log seperti berikut dihasilkan:

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