DataWorks:Alur kerja yang dipicu

Langkah 1: Membuat alur kerja yang dipicu

1. Buka halaman Workspaces di Konsol DataWorks. Di bilah navigasi atas, pilih wilayah yang diinginkan. Temukan ruang kerja yang dimaksud dan pilih Shortcuts > Data Studio pada kolom Actions.

2. Di panel navigasi kiri, klik . Di samping Business Flow, klik > Create Workflow untuk membuka halaman Create Workflow.

3. Di halaman Create Workflow, atur Scheduling Type menjadi Triggered Scheduling, masukkan Name alur kerja, lalu klik OK.

Langkah 2: Mengatur orkestrasi alur kerja dan mengembangkan node

1. Di bilah alat, klik + Add Node untuk membuka daftar node. Dari daftar jenis node di sebelah kiri, seret node Shell ke kanvas dan masukkan nama untuk membuatnya.

2. Klik ganda node Shell untuk membuka editor kode dan masukkan kode berikut:

   echo "Hello, Trigger Workflow! Current time is ${bizdate}"

3. Klik Save di bilah alat.

Langkah 3: Debug di lingkungan pengembangan

1. Kembali ke kanvas alur kerja dan klik ikon di bilah alat atas.

2. Di kotak dialog yang muncul, masukkan Current run value untuk alur kerja (misalnya, jika tanggal saat ini adalah 20260310, bizdate harus diganti dengan 20260309).

3. Di log eksekusi bagian bawah, Anda dapat melihat status eksekusi node dan output perintah echo.

Langkah 4: Publikasikan dan jalankan di produksi

1. Di kanvas alur kerja, klik tombol Publish dan ikuti petunjuk untuk mempublikasikan alur kerja.

2. Setelah alur kerja dipublikasikan, buka Operation Center > Manually Triggered Node O&M > Manual Task> Triggered Workflow.

3. Temukan alur kerja yang telah Anda publikasikan dan klik Run di kolom Actions.

4. Di kotak dialog yang muncul, klik Run lagi untuk memicu instans alur kerja di lingkungan produksi. Anda dapat melihat detail eksekusi ini di halaman Manual Instance.

Anda kini telah menguasai dasar-dasar penggunaan alur kerja yang dipicu. Selanjutnya, kita akan menjelajahi kemampuan berbasis event-nya yang lebih canggih.

Langkah 1: Membuat pemicu OSS

1. Buka Operation Center > Tenant Schedule Setting > Trigger Management.

2. Klik Create Trigger dan konfigurasikan parameter berikut:

   Catatan

   Untuk penjelasan parameter lengkap, lihat OSS trigger.

   • Trigger name: Masukkan nama kustom, misalnya oss_new_file_trigger.

   • Applicable workspace: Pilih ruang kerja target tempat alur kerja berada.

   • Trigger event type: Pilih Object Storage Service (OSS).

   • Trigger event: Pilih oss:ObjectCreated:PutObject atau event unggah lainnya.

   • Bucket name: Pilih bucket OSS Anda.

   • File name: Tentukan path dan format file yang akan dipantau. Wildcard didukung. Misalnya, untuk memantau direktori input/ terhadap semua file .csv, masukkan input/*.csv.

   • Role configuration: Saat pertama kali menggunakan fitur ini, Anda harus menyelesaikan one-click authorization. Pilih role yang dihasilkan bernama DataWorks-EventBridge-OSS-MNS-Role-*************.

     ************* merepresentasikan ID acak 13 digit yang memastikan nama role bersifat unik.

3. Klik OK untuk membuat pemicu.

Langkah 2: Membuat dan mengaitkan alur kerja

1. Ikuti langkah-langkah dalam Panduan cepat: Membuat alur kerja yang dipicu secara manual untuk membuat alur kerja yang dipicu bernama process_oss_file_workflow.

2. Di sisi kanan kanvas alur kerja, pada panel Properties, pilih Scheduling Policy.

3. Dari daftar drop-down Trigger, pilih oss_new_file_trigger yang baru saja Anda buat.

   

Langkah 3: Mengembangkan node dan mengurai parameter event

1. Di bilah alat, klik + Add Node untuk membuka daftar node. Dari daftar jenis node di sebelah kiri, seret node Shell ke kanvas dan masukkan nama untuk membuatnya.

2. Klik ganda node tersebut dan masukkan kode berikut untuk mendapatkan serta mencetak path file dari event pemicu.

   # Ketika alur kerja dipicu, informasi event diteruskan melalui variabel bawaan workflow.triggerMessage.
   # Anda dapat menggunakan ${workflow.triggerMessage.data.oss.object.key} untuk mendapatkan path lengkap file yang diunggah.
   
   echo "========= Start Processing OSS File ========="
   message="${workflow.triggerMessage}"
   echo "Raw Value: ${message}"
   
   # Ekstrak path file dari pesan event.
   FILE_PATH="${workflow.triggerMessage.data.oss.object.key}"
   echo "A new file has arrived: ${FILE_PATH}"
   
   # Tambahkan logika pemrosesan spesifik Anda di sini.
   
   echo "========= Finish Processing OSS File ========="

   Catatan

   ${workflow.triggerMessage} mengambil seluruh isi pesan event dalam format JSON. Anda dapat menemukan format pesan OSS spesifik di EventBridge dengan membuka Event Buses > DATAWORKS_TRIGGER_FOR_BUCKET_<OSS_Bucket_Name> > Event Tracking > Event Details.

   Contoh format pesan OSS (nilai workflow.triggerMessage)

   {
       "datacontenttype": "application/json;charset=utf-8",
       "aliyunaccountid": "1***********9",
       "data": {
           "eventVersion": "1.0",
           "responseElements": {
               "requestId": "69B1***********C0A8"
           },
           "eventSource": "acs:oss",
           "eventTime": "2026-03-11T05:40:45.000Z",
           "requestParameters": {
               "sourceIPAddress": "***********"
           },
           "eventName": "ObjectCreated:PostObject",
           "userIdentity": {
               "principalId": "1***********9"
           },
           "region": "cn-hangzhou",
           "oss": {
               "bucket": {
                   "name": "******",
                   "arn": "acs:oss:cn-hangzhou:1***********9:******",
                   "virtualBucket": "",
                   "ownerIdentity": "1***********9"
               },
               "ossSchemaVersion": "1.0",
               "object": {
                   "size": 59537,
                   "objectMeta": {
                       "mimeType": "text/csv"
                   },
                   "deltaSize": 0,
                   "eTag": "63***********D32",
                   "key": "input/***********.csv"
               }
           }
       },
       "subject": "acs:oss:cn-hangzhou:1***********9:dwoss1024/input/******.csv",
       "aliyunoriginalaccountid": "1***********9",
       "source": "acs.oss",
       "type": "oss:ObjectCreated:PostObject",
       "aliyunpublishtime": "2026-03-11T05:40:45.682Z",
       "specversion": "1.0",
       "aliyuneventbusname": "DATAWORKS_TRIGGER_FOR_BUCKET_******",
       "id": "69B1***********0A8",
       "time": "2026-03-11T05:40:45.000Z",
       "aliyunregionid": "cn-hangzhou"
   }

Langkah 4: Debug dan publikasikan

1. Debug:

   • Kembali ke kanvas alur kerja dan klik tombol Run .

   • Di kotak input Trigger Message Body, tempel event OSS simulasi dalam format JSON. Anda dapat menyalin contoh pesan dari halaman konfigurasi pemicu dan mengubah nilai field key. Berikut contoh minimal:

     {
       "data": {
         "oss":{
           "object": {
             "key": "input/test_file_20260310.csv" 
           }
         } 
       }
     }

   • Klik Run dan periksa log untuk memastikan bahwa input/test_file_20260310.csv berhasil dicetak.

2. Publish: Setelah debugging berhasil, klik tombol Publish untuk men-deploy alur kerja ke lingkungan produksi. Pemicu berbasis event hanya aktif di lingkungan produksi.

Langkah 5: Verifikasi di produksi

1. Menggunakan konsol OSS atau klien, unggah file CSV ke bucket dan path yang telah Anda konfigurasikan di pemicu, misalnya direktori input/.

   Cara memastikan event berhasil dipicu

   Buka https://eventbridge.console.alibabacloud.com/<regionId>/event-bus/DATAWORKS_TRIGGER_FOR_BUCKET_<OssBucketName>/event-tracing untuk mengkueri daftar event pemicu terbaru. Anda juga dapat mengklik Event Details untuk event tertentu guna melihat pesan pemicunya, yaitu nilai workflow.triggerMessage.

   

2. Di DataWorks, buka Operation Center > Manually Triggered Node O&M > Manual Task > Triggered Workflow. Alur kerja yang telah dipublikasikan process_oss_file_workflow muncul dalam daftar.

   

3. Setelah beberapa saat, instans alur kerja baru secara otomatis dipicu dan dijalankan. Anda dapat melihatnya di DataWorks pada Operation Center > Manually Triggered Node O&M > Manual Instance. Klik instans tersebut untuk melihat log-nya dan pastikan path file telah diproses dengan benar.

Skenario 2: Memproses pesan ApsaraMQ for Kafka

Tujuan: Pantau topik Kafka untuk log perilaku pengguna. Ketika pesan baru tiba, picu alur kerja untuk menguraikannya dan mengeksekusi logika berbeda berdasarkan isinya.

Orkestrasi alur kerja

Mengatur orkestrasi alur kerja yang dipicu mirip dengan mengatur orkestrasi alur kerja terjadwal. Untuk informasi lebih lanjut, lihat Node and workflow orchestration.

Parameter penjadwalan

Di panel Properties di sisi kanan kanvas alur kerja, Anda dapat mengatur parameter global untuk alur kerja. Semua node di dalamnya dapat mereferensikan parameter tersebut.

• Sintaks referensi: Di kode node, referensikan parameter alur kerja dengan format ${workflow.parameter_name}.

• Prioritas parameter: Parameter di DataWorks memiliki hubungan override hierarkis. Urutan prioritasnya adalah: parameter node > parameter alur kerja.

  Untuk informasi lebih lanjut tentang parameter, lihat Parameter design and flow.

Kebijakan penjadwalan

Ketika beberapa alur kerja dipicu secara bersamaan dan menyebabkan bottleneck sumber daya, gunakan priority dan priority weighting policies untuk mengelola penjadwalan sumber daya dan memastikan tugas paling penting dieksekusi terlebih dahulu.

• Menjamin kelangsungan bisnis inti: Atur prioritas lebih tinggi untuk alur kerja bisnis kritis agar selalu berjalan sebelum alur kerja non-kritis.

• Mengurangi waktu eksekusi jalur kritis: Dalam satu instans alur kerja, Anda dapat menggunakan priority weighting policy untuk memengaruhi urutan eksekusi node. Misalnya, dengan kebijakan downward weighting, node pada jalur kritis yang memiliki lebih banyak dependensi hulu mendapatkan bobot dinamis lebih tinggi. Strategi ini memprioritaskan eksekusi mereka, sehingga memperpendek waktu eksekusi keseluruhan alur kerja.

  Parameter	Description
  Priority	Menentukan tingkat prioritas absolut instans alur kerja dalam antrean penjadwalan. Tingkat yang tersedia adalah 1, 3, 5, 7, dan 8, di mana angka lebih tinggi menunjukkan prioritas lebih tinggi. Tugas atau alur kerja berprioritas tinggi selalu dialokasikan sumber daya penjadwalan sebelum yang berprioritas rendah.
  Priority weighting policy	Menentukan cara bobot masing-masing node (task) dihitung secara dinamis dalam tingkat prioritas yang sama. Node dengan bobot lebih tinggi diberi prioritas eksekusi. No weighting: Semua node memiliki bobot garis dasar tetap. Downward weighting: Bobot node disesuaikan secara dinamis. Semakin banyak node hulu yang menjadi dependensinya, semakin tinggi bobotnya. Strategi ini membantu memprioritaskan eksekusi node pada jalur kritis dalam grafik asiklik terarah (DAG). Bobot dihitung sebagai berikut: Initial weight + Sum of the priorities of all its upstream nodes.
  Maximum parallel instances	Mengontrol jumlah maksimum instans alur kerja ini yang dapat berjalan secara bersamaan untuk kontrol konkurensi dan perlindungan sumber daya. Ketika batas ini tercapai, instans baru yang dipicu masuk ke status menunggu. Anda dapat mengaturnya menjadi Unlimited atau nilai kustom hingga 100.000. Catatan Jika batas yang ditentukan melebihi kapasitas maksimum kelompok sumber daya, konkurensi aktual ditentukan oleh batas fisik kelompok sumber daya tersebut.

Sistem prioritas DataWorks mengikuti aturan override hierarkis: runtime specification > node-level configuration > workflow-level configuration.

1. Konfigurasi tingkat alur kerja (garis dasar): Dikonfigurasi di panel Scheduling Policy alur kerja, ini berfungsi sebagai pengaturan default untuk semua node.

2. Konfigurasi tingkat node (override): Anda dapat meng-override pengaturan tingkat alur kerja untuk node tertentu dengan mengatur priority lebih tinggi di panel Properties > Scheduling Policy-nya.

3. Spesifikasi runtime (sementara): Saat memicu eksekusi secara manual di Operation Center, Anda dapat menentukan prioritas dengan menggunakan sakelar Override Priority at Runtime. Konfigurasi ini memiliki prioritas tertinggi dan hanya berlaku untuk eksekusi saat ini, tanpa mengubah pengaturan permanen apa pun.