DataWorks:Alur Kerja yang Dipicu

Langkah 1: Buat alur kerja yang dipicu

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

2. Di bilah navigasi kiri, klik , lalu di sebelah kanan Project Folder, klik > New Workflow untuk membuka halaman New Workflow.

3. Pada kotak dialog yang muncul, di halaman Create Workflow, atur Scheduling Type ke Trigger-based Scheduling. Masukkan Name alur kerja, lalu klik Confirm.

Langkah 2: Orkestrasi dan kembangkan node

1. Klik + Add Node di bilah alat untuk membuka daftar node. Seret node Shell dari daftar tipe node ke kanvas dan masukkan nama.

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 dan jalankan (lingkungan pengembangan)

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

2. Pada kotak dialog yang muncul, masukkan Value for This Run untuk alur kerja (misalnya, jika tanggal saat ini adalah 20260310, nilai untuk bizdate harus 20260309).

3. Tak lama kemudian, di log waktu proses di bawah, Anda dapat melihat status berjalan node dan output perintah echo.

Langkah 4: Publikasikan dan jalankan (lingkungan produksi)

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

2. Setelah alur kerja dipublikasikan, buka Operation Center > One-time Task O&M > One-time Task > Triggered Workflow.

3. Temukan alur kerja yang telah dipublikasikan, lalu klik Run di kolom Operation.

4. Pada jendela yang muncul, klik Run lagi untuk memicu instans alur kerja di Lingkungan Produksi. Anda dapat melihat detail eksekusi ini di halaman One-time Instance.

Anda kini telah mempelajari dasar-dasar Triggered Workflow. Selanjutnya, panduan ini akan menjelajahi kemampuan pemicu berbasis event yang lebih canggih.

Langkah 1: Buat pemicu OSS

1. Buka Operation Center > Scheduling Settings > Trigger Management.

2. Klik New Trigger dan konfigurasikan sebagai berikut:

   Catatan

   Untuk deskripsi parameter lengkap, lihat OSS Trigger.

   • Trigger Name: Masukkan nama kustom, misalnya oss_new_file_trigger.

   • Applicable workspace: Pilih ruang kerja target tempat alur kerja Anda 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, Anda dapat memasukkan input/*.csv.

   • Role Configuration: Untuk penggunaan pertama kali, lakukan One-click Authorization dan pilih role yang dihasilkan bernama DataWorks-EventBridge-OSS-MNS-Role-*************.

     ************* merepresentasikan ID acak 13 digit yang digunakan untuk memastikan keunikan.

3. Klik Confirm untuk membuat Trigger.

Langkah 2: Buat dan kaitkan alur kerja

1. Ikuti langkah-langkah di Panduan Cepat: Buat Alur Kerja yang Dipicu Manual untuk membuat Triggered Workflow bernama process_oss_file_workflow.

2. Di panel kanan kanvas alur kerja, pilih Schedule Configuration > Scheduling Policy.

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

   

Langkah 3: Uraikan parameter event

1. Klik + Add Node di bilah alat untuk membuka daftar node. Seret node Shell dari daftar tipe node ke kanvas dan masukkan nama untuk membuatnya.

2. Klik ganda node tersebut dan tulis kode untuk mengambil serta mencetak path file dari event pemicu.

   # Saat Trigger memulai alur kerja, informasi event diteruskan melalui variabel bawaan workflow.triggerMessage.
   # Anda dapat mengakses path lengkap file yang diunggah menggunakan ${workflow.triggerMessage.data.oss.object.key}.
   
   echo "========= Start Processing OSS File ========="
   message="${workflow.triggerMessage}"
   echo "Raw Value: ${message}"
   
   # Ekstrak nama 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 isi pesan event lengkap dalam format JSON. Anda dapat memperoleh format pesan spesifik untuk OSS di EventBridge di bawah Event Bus > DATAWORKS_TRIGGER_FOR_BUCKET_<OSS_Bucket_Name> > Event Tracing > Event Details.

   Contoh format pesan OSS

   {
       "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 contoh event OSS dalam format JSON. Anda dapat menyalin contoh format pesan dari halaman konfigurasi trigger dan mengubah nilai key. Berikut contoh sederhananya.

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

   • Klik Run, lalu periksa apakah input/test_file_20260310.csv berhasil dicetak di log.

2. Publish: Setelah debugging berhasil, klik tombol Publish untuk mempublikasikan alur kerja ke Lingkungan Produksi. Pemicu berbasis event hanya berlaku di Lingkungan Produksi.

Langkah 5: Verifikasi di produksi

1. Dengan menggunakan konsol OSS atau klien, unggah file CSV ke bucket dan path yang telah Anda konfigurasikan di trigger (misalnya, direktori input/).

   Verifikasi pemicu event

   Buka https://eventbridge.console.alibabacloud.com/<regionId>/event-bus/DATAWORKS_TRIGGER_FOR_BUCKET_<OssBucketName>/event-tracing untuk melihat daftar event yang baru dipicu. Anda juga dapat mengklik Event Details untuk melihat pesan yang dipicu secara spesifik (workflow.triggerMessage).

   

2. Buka DataWorks Operation Center > One-time Task Operations > One-time Task > Triggered Workflow. Alur kerja process_oss_file_workflow yang berhasil dipublikasikan akan ditampilkan.

   

3. Setelah menunggu sebentar, buka DataWorks Operation Center > One-time Task O&M > Triggered Workflow Instances. Instans alur kerja baru akan dipicu secara otomatis. Klik untuk melihat log-nya dan pastikan path file diproses dengan benar.

Skenario 2: Memproses pesan Kafka

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

Orkestrasi alur kerja

Proses inti orkestrasi Triggered Workflow mirip dengan Recurring Workflow. Untuk informasi selengkapnya, lihat Orchestrate nodes and workflows.

Parameter penjadwalan

Di panel Schedule Configuration di sisi kanan kanvas alur kerja, Anda dapat mengatur parameter global untuk workflow. Semua node di dalamnya dapat mereferensikan parameter ini.

• 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 selengkapnya tentang parameter, lihat Parameter design and flow.

Kebijakan penjadwalan

Saat beberapa alur kerja atau tugas dipicu secara bersamaan hingga menyebabkan bottleneck sumber daya sistem, Anda dapat menggunakan Priority dan Priority Weighting Policy untuk menerapkan penjadwalan sumber daya cerdas. Hal ini memastikan tugas paling penting dieksekusi terlebih dahulu.

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

• Mengurangi durasi 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 akan diberi bobot dinamis lebih tinggi. Hal ini memprioritaskan eksekusinya, sehingga dapat memperpendek waktu proses alur kerja secara keseluruhan.

  version	versi
  Priority	Menentukan tingkat prioritas absolut suatu instans alur kerja dalam antrian 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 menerima sumber daya penjadwalan sebelum yang berprioritas rendah.
  Priority Weighting Policy	Menentukan cara bobot node internal (tugas) dihitung secara dinamis dalam tingkat prioritas yang sama. Node dengan bobot lebih tinggi diprioritaskan untuk dieksekusi. Tanpa pembobotan: Semua node memiliki bobot garis dasar tetap. Downward Weighting: Bobot node disesuaikan secara dinamis. Semakin banyak dependensi hulu yang dimiliki node, semakin tinggi bobotnya. Strategi ini membantu memprioritaskan eksekusi node pada jalur kritis dalam DAG (Directed Acyclic Graph). Bobot dihitung sebagai berikut: Bobot awal + Jumlah prioritas semua node hulunya.
  Maximum Parallel Instances	Mengontrol jumlah maksimum instans alur kerja yang dapat berjalan secara konkuren. Ini digunakan untuk kontrol konkurensi dan perlindungan sumber daya. Saat jumlah instans yang sedang berjalan mencapai batas, instans baru yang dipicu akan masuk ke status menunggu. Anda dapat mengatur nilai ini ke Unlimited atau menentukan nilai maksimum kustom hingga 100.000. Catatan Jika batas yang ditentukan melebihi kapasitas maksimum kelompok sumber daya, batas fisik kelompok sumber daya tersebut menjadi bottleneck konkurensi aktual.

Sistem prioritas DataWorks mengikuti aturan override hierarkis: Spesifikasi runtime > Konfigurasi tingkat node > Konfigurasi tingkat alur kerja.

1. Konfigurasi tingkat alur kerja (Garis dasar): Diatur di Scheduling Policy alur kerja, berfungsi sebagai pengaturan default untuk semua node.

2. Konfigurasi tingkat node (Lokal): Di Schedule Configuration > Scheduling Policy node individual dalam alur kerja, Anda dapat mengatur Priority lebih tinggi untuk node tersebut, yang menggantikan pengaturan tingkat alur kerja.

3. Spesifikasi runtime (Sementara): Saat memicu eksekusi manual di Operation Center, Anda dapat menentukan konfigurasi menggunakan sakelar Reset Priority At Runtime. Konfigurasi ini memiliki prioritas tertinggi, hanya berlaku untuk eksekusi saat ini, dan tidak mengubah pengaturan permanen apa pun.