DataWorks:Pengembangan Notebook Tingkat Lanjut

Buat dan publikasikan resource Python

1. Di panel navigasi kiri DataWorks DataStudio, klik untuk membuka Resource Management.

2. Klik kanan direktori target atau klik ikon + di pojok kanan atas. Pilih Create Resource > MaxCompute Python, lalu beri nama file my_utils.py.

3. Di bagian File Content, klik Edit Online, tempel kode fungsi utilitas Anda, lalu klik Save.

   # my_utils.py
   def greet(name):
       return f"Hello, {name} from resource file!"

4. Klik Save, lalu Publish di bilah alat. Resource tersebut menjadi tersedia untuk tugas pengembangan maupun produksi.

Referensi resource di Notebook

Di baris pertama sel Python, gunakan ##@resource_reference untuk mereferensikan resource yang telah dipublikasikan:

##@resource_reference{"my_utils.py"}
# Jika resource berada di subdirektori (misalnya, my_folder/my_utils.py),
# referensikan tanpa nama direktori: ##@resource_reference{"my_utils.py"}
from my_utils import greet

message = greet('DataWorks')
print(message)

Debug di lingkungan pengembangan

Jalankan sel Python tersebut. Output-nya adalah:

Hello, DataWorks from resource file!

Publikasikan ke lingkungan produksi

Setelah menyimpan dan mempublikasikan node Notebook, buka Operation Center > Recurring Tasks lalu klik Test. Setelah tugas berhasil, output Hello, DataWorks from resource file! muncul di log.

Untuk informasi lebih lanjut, lihat Resource dan fungsi MaxCompute.

Debug di lingkungan pengembangan

1. Pasang dataset: Di halaman detail lingkungan pengembangan pribadi Anda, konfigurasikan dataset di bawah Storage Configuration > Dataset.

2. Akses dalam kode: Dataset dipasang ke path di lingkungan pengembangan pribadi. Baca atau tulis langsung ke path tersebut:

   # Asumsikan dataset dipasang ke /mnt/data/dataset.
   import pandas as pd
   
   file_path = '/mnt/data/dataset/testfile.csv'
   df = pd.read_csv(file_path)
   
   # Tulis data ke MaxCompute menggunakan PyODPS.
   o = %odps
   o.write_table('mc_test_table', df, overwrite=True)
   print(f"Data berhasil ditulis ke tabel MaxCompute mc_test_table.")

Deploy ke produksi

1. Pasang dataset: Di halaman pengeditan node Notebook, tambahkan dataset di bawah scheduling configurations > scheduling policy di panel navigasi kanan.

2. Akses dalam kode: Setelah melakukan commit dan mempublikasikan node, dataset dipasang ke path di lingkungan produksi. Gunakan path yang sama dalam kode Anda:

   # Asumsikan dataset dipasang ke /mnt/data/dataset.
   import pandas as pd
   
   file_path = '/mnt/data/dataset/testfile.csv'
   df = pd.read_csv(file_path)
   
   # Tulis data ke MaxCompute menggunakan PyODPS.
   o = %odps
   o.write_table('mc_test_table', df, overwrite=True)
   print(f"Data berhasil ditulis ke tabel MaxCompute mc_test_table.")

Untuk informasi lebih lanjut, lihat Gunakan dataset di lingkungan pengembangan pribadi.

%odps — Dapatkan objek entri PyODPS

Mengembalikan objek PyODPS terautentikasi yang terikat ke proyek MaxCompute saat ini. Ini menghindari hard-coding AccessKey dalam kode Anda.

o = %odps

Setelah menjalankan perintah, pemilih resource komputasi muncul di pojok kanan bawah dengan proyek yang dipilih secara otomatis. Klik nama proyek untuk mengganti proyek.

Gunakan objek tersebut untuk menjalankan skrip PyODPS. Misalnya, untuk menampilkan semua tabel di proyek saat ini:

with o.execute_sql('show tables').open_reader() as reader:
    print(reader.raw)

%maxframe — Buat koneksi MaxFrame

Membuat session MaxFrame untuk pemrosesan data terdistribusi bergaya pandas pada data MaxCompute.

# Hubungkan ke MaxCompute menggunakan MaxFrame.
mf_session = %maxframe

df = mf_session.read_odps_table('your_mc_table')
print(df.head())

# Hancurkan session secara manual untuk melepas resource setelah debugging.
mf_session.destroy()

Perbandingan mesin

Setelah terhubung ke mesin Spark, konteks eksekusi seluruh kernel Notebook beralih ke lingkungan PySpark remote. Tulis kode PySpark langsung di sel-sel berikutnya.

MaxCompute Spark

Bind a MaxCompute compute resource sebelum membuat koneksi.

Hubungkan melalui Livy ke mesin Spark yang tertanam di proyek MaxCompute Anda.

1. Buat koneksi: Jalankan perintah berikut di sel Python. Sistem secara otomatis membuat atau menggunakan ulang session Spark.

   # Buat session Spark dan mulai Livy.
   %maxcompute_spark

2. Eksekusi kode PySpark: Gunakan magic sel %%spark di sel Python baru.

   # Sel Python yang menggunakan MaxCompute Spark harus dimulai dengan %%spark.
   %%spark
   
   df = spark.sql("SELECT * FROM your_mc_table LIMIT 10")
   df.show()

3. Lepaskan koneksi: Setelah debugging, hentikan atau hapus session. Di lingkungan produksi, sistem secara otomatis menghentikan dan menghapus session Livy ketika instans tugas berakhir.

   # Hentikan session Spark dan Livy.
   %maxcompute_spark stop
   
   # Hentikan Livy dan hapus.
   %maxcompute_spark delete

EMR Serverless Spark

Bind an EMR Serverless Spark compute resource ke ruang kerja Anda dan buat Livy Gateway sebelum membuat koneksi.

Berinteraksi dengan EMR Serverless Spark dengan menghubungkan ke Livy Gateway yang telah dibuat sebelumnya.

1. Buat koneksi: Pilih EMR compute resource dan Livy gateway di pojok kanan bawah sel, lalu jalankan salah satu perintah berikut:

   • Dipilih: konfigurasi global menggantikan parameter kustom session.

   • Tidak dipilih: parameter kustom session menggantikan konfigurasi global.

   # Koneksi dasar
   %emr_serverless_spark

   Untuk mengirimkan parameter Spark kustom, gunakan %%emr_serverless_spark (dua tanda persen):

   %%emr_serverless_spark
   {
     "spark_conf": {
       "spark.emr.serverless.environmentId": "<EMR_Serverless_Spark_Runtime_Environment_ID>",
       "spark.emr.serverless.network.service.name": "<EMR_Serverless_Spark_Network_Connection_ID>",
       "spark.driver.cores": "1",
       "spark.driver.memory": "8g",
       "spark.executor.cores": "1",
       "spark.executor.memory": "2g",
       "spark.driver.maxResultSize": "32g"
     }
   }

   Parameter kustom hanya berlaku untuk session saat ini. Jika diabaikan, sistem menggunakan parameter global yang dikonfigurasi di Admin Center. Untuk berbagi konfigurasi lintas tugas dan pengguna, atur secara global di Admin Center > Serverless Spark > SPARK parameters. Opsi Prioritize Global Configurations di Admin Center mengontrol prioritas ketika parameter yang sama muncul di kedua tempat:

2. (Opsional) Sambung ulang: Jika administrator menghapus token di halaman Livy gateway, buat ulang dengan:

   # Sambung ulang dan segarkan token Livy.
   %emr_serverless_spark refresh_token

3. Eksekusi kode PySpark atau SQL: Setelah koneksi berhasil, kernel beralih. Tulis kode PySpark langsung di sel Python, atau tulis SQL di sel EMR Spark SQL.

   • Kirim SQL ke EMR Serverless Spark menggunakan sel EMR Spark SQL — sel ini menggunakan ulang koneksi dari %emr_serverless_spark dan mengirimkan job secara otomatis. Tidak perlu memilih resource komputasi di dalam sel.

   • Kirim kode PySpark di sel Python — tidak perlu awalan %%spark.

4. Lepaskan koneksi:

   Penting

   Jika beberapa pengguna berbagi Livy Gateway, perintah stop atau delete memengaruhi semua pengguna yang sedang menggunakan gateway tersebut. Gunakan perintah ini dengan hati-hati.

   # Hentikan session Spark dan Livy.
   %emr_serverless_spark stop
   
   # Hentikan Livy dan hapus.
   %emr_serverless_spark delete

AnalyticDB for Spark

Bind an AnalyticDB for Spark compute resource ke ruang kerja Anda sebelum membuat koneksi.

Hubungkan ke mesin AnalyticDB for Spark dengan membuat Spark Connect Server.

1. Buat koneksi: Pilih ADB Spark compute resource di pojok kanan bawah sel. Konfigurasikan ID vSwitch dan ID Security Group untuk memastikan konektivitas jaringan, lalu jalankan:

   • ID vSwitch (vswitchId): Di konsol Alibaba Cloud AnalyticDB MySQL, lihat vSwitch ID di Network Information pada halaman detail instans.

   • ID Security Group (securityGroupId): Di Network Settings pada halaman detail lingkungan pengembangan pribadi Anda, temukan ID Security Group yang dipilih — ID tersebut diawali dengan sg-.

   Penting

   Untuk memastikan konektivitas jaringan, pilih Virtual Private Cloud (VPC) dan vSwitch yang sama dengan instans AnalyticDB for Spark Anda saat membuat lingkungan pengembangan pribadi.

   # Konfigurasikan ID vSwitch dan ID Security Group untuk konektivitas jaringan.
   %adb_spark add  --spark-conf spark.adb.version=3.5  --spark-conf spark.adb.eni.enabled=true  --spark-conf spark.adb.eni.vswitchId=<vSwitch_ID_of_ADB>  --spark-conf spark.adb.eni.securityGroupId=<Security_Group_ID_of_personal_development_environment>

   Cara menemukan ID vSwitch dan Security Group:

2. Eksekusi kode PySpark: Setelah koneksi terbentuk, jalankan PySpark di sel Python baru.

   Mesin AnalyticDB for Spark hanya dapat memproses tabel C-Store yang memiliki atribut 'storagePolicy'='COLD'.

   # AnalyticDB for Spark hanya dapat memproses tabel C-Store.
   df = spark.sql("SELECT * FROM my_adb_cstore_table LIMIT 10")
   df.show()

3. Lepaskan koneksi: Setelah debugging, bersihkan session koneksi untuk menghemat resource. Di produksi, sistem membersihkan secara otomatis.

   %adb_spark cleanup

Buat koneksi

Jalankan %lindorm_ray di sel Python. Pemilih resource komputasi muncul di pojok kanan bawah — pilih Lindorm compute resource dan grup resource RAY yang telah dibuat.

# Hubungkan ke grup resource Lindorm Ray yang ditentukan.
%lindorm_ray

Eksekusi kode Ray

Setelah koneksi berhasil, tulis dan eksekusi kode Ray di sel Python baru. Log dialirkan kembali ke area output sel secara real time.

Contoh berikut mendefinisikan tugas remote menggunakan dekorator @ray.remote. Tugas tersebut dijalankan di kluster Ray, dan log serta hasilnya dikembalikan ke area output:

import ray
import time

@ray.remote
def hello_world():
  print("Hello from Lindorm Ray!")
  time.sleep(5)
  return "Task finished."

# Kirim tugas remote.
result_ref = hello_world.remote()
print(ray.get(result_ref))

Tentukan parameter startup kustom (opsional)

Untuk menginstal paket Python pihak ketiga atau mengunggah file kode lokal, gunakan %%lindorm_ray untuk membuat koneksi dengan konfigurasi lingkungan runtime kustom.

Contoh 1: Instal dependensi

Instal paket jieba di lingkungan Ray menggunakan parameter pip:

%%lindorm_ray
{
  "runtime_env": {
    "pip": ["jieba"]
  }
}

Setelah lingkungan siap, impor dan gunakan paket tersebut di tugas Ray berikutnya:

import ray

@ray.remote
def do_work(x):
    import jieba
    return "/".join(jieba.cut(x))

print(ray.get(do_work.remote("Welcome to the DataWorks+LindormRay solution")))

Contoh 2: Unggah dan gunakan resource DataWorks

Gunakan parameter working_dir untuk mengunggah resource dari DataWorks Resource Management ke kluster Ray:

# Referensi resource dari DataStudio Resource Management dan deklarasikan path-nya.
%%lindorm_ray
{
    "runtime_env": {
        "working_dir": "/mnt/workspace/_dataworks/resource_references"
    }
}

Asumsikan ray_resource.py telah diunggah ke DataStudio Resource Management:

ray_resource.py:

def fun():
    print("This is a test function in ray_resource.py")

Referensi dan gunakan dalam tugas Ray:

import ray

##@resource_reference{"ray_resource.py"}

@ray.remote
def do_work(x):
    print('Ray says:', x)
    from ray_resource import fun
    fun()
    return x

worker = do_work.remote("Welcome to the DataWorks+LindormRay solution")
print(ray.get(worker))

Penjadwalan produksi dan O&M

Setelah pengembangan dan debugging selesai, lakukan commit dan publikasikan node Notebook. Node tersebut dijalankan sebagai node Lindorm Ray dalam DAG sesuai jadwal periodik.

• Parameterisasi: Gunakan parameter penjadwalan DataWorks standar seperti ${bizdate}.

• Penampilan log: Untuk mencegah log berlebihan memengaruhi kinerja, sistem hanya memuat 1 MB log pertama secara default. Jika log terpotong, output mencakup tautan ke konsol Lindorm untuk log tugas lengkap.

• Pelepasan resource: Setelah tugas produksi terjadwal selesai, tugas Lindorm Ray memasuki status yang diinginkan dan melepas resource. Selama pengembangan interaktif, restart kernel atau tutup Notebook untuk melepas resource.