DataWorks:Alur kerja yang dipicu

Langkah 1: Membuat alur kerja yang dipicu

1. Buka halaman Ruang Kerja di konsol DataWorks. Di bilah navigasi atas, pilih wilayah yang diinginkan. Temukan ruang kerja yang diinginkan dan pilih Shortcuts > Data Studio di kolom Actions.

2. Di panel navigasi kiri, klik , lalu klik > Create Workflow di sebelah kanan Project Directory untuk membuka halaman Create Workflow.

3. Pada kotak dialog yang muncul, di halaman Create Workflow, atur Scheduling Type menjadi Triggered scheduling. Masukkan Name alur kerja dan klik Confirm untuk membuat alur kerja.

Langkah 2: Mengatur orkestrasi alur kerja dan mengembangkan node

1. Klik + Add Node di bilah alat untuk membuka daftar node. Seret node Shell dari daftar jenis node di sebelah kiri ke kanvas, masukkan nama, lalu selesaikan pembuatan.

2. Klik ganda node Shell untuk membuka halaman pengeditan kode, lalu masukkan kode berikut:

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

3. Klik tombol 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 Used in This Run untuk alur kerja tersebut (misalnya, jika hari ini 20260310, bizdate harus diganti dengan 20260309).

3. Setelah beberapa saat, di log eksekusi bagian bawah, Anda dapat melihat status eksekusi node dan output perintah echo.

Langkah 4: Terapkan dan jalankan (lingkungan produksi)

1. Di kanvas alur kerja, klik tombol Deploy dan ikuti petunjuk untuk menyelesaikan proses penerapan.

2. Setelah penerapan berhasil, buka Operation Center > Manually Triggered Task O&M > Manually Triggered Task > Triggered Workflow.

3. Temukan alur kerja yang baru saja Anda terapkan, lalu klik Run di kolom Operation.

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

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

Langkah 1: Membuat pemicu OSS

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

2. Klik Create Trigger dan konfigurasikan sebagai berikut:

   Catatan

   Untuk deskripsi parameter lengkap, lihat Pemicu OSS.

   • 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 semua file .csv di direktori input/, masukkan input/*.csv.

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

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

3. Klik Confirm untuk menyelesaikan pembuatan 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 baru bernama process_oss_file_workflow.

2. Di panel kanan kanvas alur kerja, navigasikan ke Scheduling Settings > Scheduling Policy.

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

   Setelah memilih pemicu, Anda dapat menggunakan ${workflow.triggerMessage} di tugas internal untuk mendapatkan isi pesan lengkap, atau gunakan ${workflow.triggerMessage.xxx} untuk mendapatkan nilai bidang tertentu dalam isi pesan.

Langkah 3: Mengembangkan node dan mengurai parameter event

1. Klik + Add Node di bilah alat untuk membuka daftar node. Seret node Shell dari daftar jenis node di sebelah kiri ke kanvas, masukkan nama, lalu selesaikan pembuatan.

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

   1. Penggunaan umum: Penggunaan parameter berikut berlaku untuk sebagian besar pesan event.

      # Saat pemicu menjalankan alur kerja, info event diteruskan melalui variabel bawaan workflow.triggerMessage
      # Kita bisa mendapatkan path lengkap file yang diunggah melalui ${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 di sini
      echo "========= Finish Processing OSS File ========="

   2. Kasus khusus: Beberapa node, seperti node Shell, memiliki aturan sintaks sendiri. Sebelum node dijalankan, mesin penjadwalan terlebih dahulu mengganti konten pesan ke dalam teks skrip, lalu meneruskannya ke interpreter Shell untuk dieksekusi. Akibatnya, tanda kutip, spasi, dan jeda baris dalam pesan langsung berpartisipasi dalam parsing sintaks Shell. Hal ini dapat menyebabkan spasi atau jeda baris dalam parameter merusak struktur kode asli setelah node mengambil parameter tersebut. Dalam kasus ini, Anda dapat menggunakan heredoc untuk membaca pesan, yang menghindari sebagian besar masalah karakter khusus:

      # Gunakan heredoc untuk menangkap teks mentah yang disubstitusi dari ${workflow.triggerMessage}.
      message=$(cat <<'__DW_TRIGGER_MESSAGE__'
      ${workflow.triggerMessage}
      __DW_TRIGGER_MESSAGE__
      )
      
      # Skenario 1: Mendapatkan seluruh pesan mentah
      printf 'Raw Value: %s\n' "$message"
      
      # Skenario 2: Jika hanya membutuhkan bidang tertentu, gunakan Python untuk mengurai JSON dan mengekstrak nilainya
      dt=$(printf '%s' "$message" | /home/tops/bin/python3 -c 'import json,sys; print(json.load(sys.stdin).get("body",{}).get("dt",""))')
      printf 'dt=%s\n' "$dt"

      Catatan

      ${workflow.triggerMessage}: Mengambil isi pesan event lengkap dalam format JSON. Anda dapat memperoleh format pesan spesifik untuk OSS dari EventBridge > Event Bus > DATAWORKS_TRIGGER_FOR_BUCKET_<OSS_Bucket_Name> > Event Tracing > Event Details.

      Lihat contoh format pesan OSS, yaitu format dari 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 JSON event OSS simulasi. Anda dapat menyalin dan memodifikasi nilai key dari "Contoh Format Pesan" di halaman konfigurasi pemicu. Berikut contoh sederhananya.

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

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

2. Publikasi: Setelah debug berhasil, klik tombol Publish untuk menerapkan alur kerja ke lingkungan produksi. Pemicu event hanya berlaku di lingkungan produksi.

Langkah 5: Verifikasi di produksi

1. Gunakan konsol OSS atau alat klien untuk mengunggah file CSV ke bucket dan path (misalnya direktori input/) yang Anda konfigurasikan di pemicu.

   Bagaimana cara memverifikasi bahwa pemicu event berfungsi?

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

   Daftar event menampilkan catatan pemicu dengan sumber event Object Storage Service (OSS) / acs.oss dan jenis event oss:ObjectCreated:PostObject.

2. Buka DataWorks Operation Center > Manually Triggered Task O&M > Manually Triggered Task > Triggered Workflow. Alur kerja process_oss_file_workflow yang berhasil dipublikasikan akan muncul.

3. Tunggu sejenak, lalu buka DataWorks Operation Center > Manually Triggered Task O&M > Triggered Workflow Instance. Instans alur kerja baru akan dipicu secara otomatis. Klik untuk melihat log-nya dan verifikasi bahwa path file diproses dengan benar.

   ===== Start Processing 0SS File =====
   Raw Value: {"datacontenttype":"application/json;charset=utf-8","aliyunaccountid":"1162423445433459","data":{"eventVersion":"1.0","responseElements":{"requestId":"69B1083F7A439F343040ABCD"},
   "eventSource":"acs:oss","eventTime":"2026-03-11T06:14:23.000Z","requestParameters":{"sourceIPAddress":"140.205.11.13"},"eventName":"0bjectCreated:Post0bject","userIdentity":{"principalId":"1162423445433459"},
   "region":"cn-hangzhou","oss":{"bucket":{"name":"dwoss1024","arn":"acs:oss:cn-hangzhou:1162423445433459:dwoss1024","virtualBucket":"","ownerIdentity":"1162423445433459"},"ossSchemaVersion":"1.0","object":{"size":59537,"objectMeta":
   {"mimeType":"text/csv"},"deltaSize":0,"eTag":"63B4BA5A45AEFC679B9A917E8DDF0D32","key":"input/2013-2020-global-PS4-game-sales.csv"}}},"subject":"acs:oss:cn-hangzhou:1162423445433459:dwoss1024/input/2013-2020-global-PS4-game-sales.csv","aliyunoriginalaccountid":"1162423445433459","source":"acs.oss","type":"oss:0bjectCreated:Post0bject","aliyunpublishtime":"2026-03-11T06:14:23.959Z","specversion":"1.0","aliyuneventbusname":"DATAW0RKS_TRIGGER_F0R_BUCKET_dwoss1024","id":"69B1083F7A439F343040ABCD","time":"2026-03-11T06:14:23.000Z","aliyunregionid":"cn-hangzhou"}A new file has arrived: input/2013-2020-global-PS4-game-sales.csv
   ====== Finish Processing 0SS File =========

Skenario 2: Kedatangan pesan Kafka mendorong komputasi real-time

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

Orkestrasi alur kerja

Proses orkestrasi inti untuk alur kerja yang dipicu mirip dengan alur kerja terjadwal. Untuk informasi lebih lanjut, lihat Orkestrasi node/alur kerja.

Parameter penjadwalan

Di panel Scheduling Settings di sisi kanan kanvas alur kerja, Anda dapat menetapkan parameter global untuk alur kerja. Semua node dalam alur kerja dapat mereferensikan parameter ini.

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

• Prioritas parameter: Parameter di DataWorks mengikuti aturan penggantian hierarkis. Urutan prioritasnya adalah: Parameter node > Parameter alur kerja.

  Untuk informasi lebih lanjut tentang parameter, lihat Desain dan alur parameter.

Kebijakan penjadwalan

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

• Lindungi bisnis inti: Tetapkan prioritas lebih tinggi untuk alur kerja bisnis inti agar selalu dijalankan sebelum alur kerja non-inti lainnya.

• Kurangi durasi jalur kritis: Dalam instans alur kerja yang sama, dengan menggunakan Priority weighting strategy, Anda dapat memengaruhi urutan eksekusi node. Misalnya, dengan strategi Downstream weighting, node pada jalur kritis yang memiliki lebih banyak dependensi hulu menerima bobot dinamis lebih tinggi dan dieksekusi terlebih dahulu, sehingga secara efektif mengurangi durasi eksekusi alur kerja secara keseluruhan.

  Item konfigurasi	Deskripsi
  Priority	Menentukan tingkat prioritas absolut instans alur kerja dalam antrian penjadwalan. Tingkat yang tersedia adalah 1, 3, 5, 7, dan 8 (semakin besar angkanya, semakin tinggi prioritasnya). Tugas atau alur kerja dengan prioritas lebih tinggi selalu mendapatkan sumber daya penjadwalan sebelum yang berprioritas lebih rendah.
  Priority weighting strategy	Menentukan cara menghitung bobot dinamis setiap node (tugas) dalam alur kerja pada tingkat prioritas yang sama. Node dengan bobot lebih tinggi dieksekusi terlebih dahulu. Tanpa pembobotan: Semua node memiliki bobot garis dasar tetap yang sama. Pembobotan downstream: Bobot node disesuaikan secara dinamis—semakin banyak dependensi upstream yang dimiliki suatu node, semakin tinggi bobotnya. Strategi ini memprioritaskan eksekusi node pada jalur kritis dalam Grafik asiklik terarah (DAG). Bobot dihitung sebagai: nilai bobot awal + jumlah prioritas dari semua node upstream.
  Maximum Parallel Instances for Internal Tasks	Mengontrol jumlah maksimum instans alur kerja ini 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 selanjutnya akan masuk ke status menunggu. Anda dapat mengaturnya ke Allowed atau menentukan nilai maksimum kustom (hingga 100.000). Catatan Jika Anda mengatur batas melebihi kapasitas maksimum kelompok sumber daya, bottleneck konkurensi aktual ditentukan oleh batas fisik kelompok sumber daya tersebut.

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

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

2. Konfigurasi tingkat node (lokal): Di Scheduling Settings > Scheduling Policy node individual dalam alur kerja, Anda dapat menetapkan Priority lebih tinggi untuk node tertentu, yang menggantikan pengaturan tingkat alur kerja.

3. Spesifikasi runtime (sementara): Ditentukan melalui toggle Runtime Priority Reset saat Anda memicu eksekusi secara manual di Operation and Maintenance Center. Konfigurasi ini memiliki prioritas tertinggi, hanya berlaku untuk eksekusi saat ini, dan tidak mengubah konfigurasi permanen apa pun.