DataWorks:Node Penugasan

Membuat dan menggunakan node penetapan di konsol

Contoh berikut menunjukkan prosedur umum dengan meneruskan hasil node assignment ke node Shell. Dalam praktiknya, jenis node apa pun dapat berfungsi sebagai node hilir.

1. Konfigurasikan node assignment hulu

   Di alur kerja target, buat dan edit node assignment. Pilih MaxCompute SQL, Python 2, atau Shell, lalu tulis kode yang menghasilkan output yang ingin Anda teruskan ke node hilir.

   print '10,20,30,40'

2. Konfigurasikan node Shell hilir

   Buat node Shell. Di halaman pengeditan node Shell, referensikan hasil hulu:

   1. Di panel Scheduling Settings di sisi kanan, pilih tab Node Context Parameters.

   2. Di bagian Input Parameters, klik Add parameters.

   3. Di dialog yang muncul, atur parameter output node hulu ke parameter outputs dari node assignment yang dikonfigurasi pada langkah sebelumnya, dan tentukan Parameter Name kustom untuk parameter input node saat ini (misalnya, param).

      Catatan

      Setelah dikonfigurasi, node hilir secara otomatis membuat dependensi terhadap node assignment hulu.

   4. Setelah konfigurasi parameter, Anda dapat menggunakan nilai yang diteruskan dari node hulu dalam kode node Shell hilir dengan format ${param}.

3. Konfigurasikan node Data Integration hilir

   Dalam praktiknya, node sinkronisasi offline Data Integration juga dapat berfungsi sebagai node hilir dari node assignment:

   1. Buat node sinkronisasi offline Data Integration dan hubungkan sebagai node hilir dari node assignment.

   2. Di halaman konfigurasi penjadwalan node sinkronisasi offline, klik tab Node Context Parameters.

   3. Klik Add parameters untuk menambahkan parameter input, lalu atur sumber nilainya ke parameter outputs dari node assignment hulu.

   4. Referensikan parameter tersebut dalam kode node: apit nilai dengan tanda kutip ganda ("") jika parameter bertipe string. Jangan tambahkan tanda kutip jika parameternya bertipe non-string.

4. Verifikasi hasil

   1. Kembali ke alur kerja dan klik Deploy di bilah alat. Pilih deployment penuh.

   2. Buka halaman Node O&M > Auto Triggered Task O&M > Auto Triggered Task di Pusat Operasi, lalu lakukan Pengujian asap.

   3. Di instans pengujian, periksa apakah hasil akhir sesuai harapan Anda.

Buat node assignment menggunakan OpenAPI

Anda juga dapat membuat node assignment dengan memanggil operasi CreateNode DataWorks OpenAPI.

Untuk membuat node di alur kerja yang sudah ada, panggil ListWorkflowDefinitions untuk mengkueri alur kerja tersebut. Lalu, gunakan Id tingkat atas dari objek alur kerja dalam respons sebagai ContainerId. Jangan gunakan Script.Id.

Konfigurasikan parameter permintaan berikut saat memanggil CreateNode:

• ProjectId: ID workspace DataWorks.

• Scene: atur parameter ini ke DATAWORKS_PROJECT saat membuat node di proyek DataWorks.

• ContainerId: Id tingkat atas dari alur kerja target. Tentukan parameter ini hanya jika Anda ingin membuat node di alur kerja yang sudah ada.

• Spec: string JSON FlowSpec yang telah diserialisasi.

Contoh FlowSpec berikut membuat node assignment MaxCompute SQL. Pastikan bahasa skrip, sumber data, dan konten skrip konsisten.

{
  "version": "1.1.0",
  "kind": "Node",
  "spec": {
    "nodes": [
      {
        "recurrence": "Normal",
        "instanceMode": "T+1",
        "datasource": {
          "name": "<MAXCOMPUTE_DATASOURCE_NAME>",
          "type": "odps"
        },
        "script": {
          "language": "odps-sql",
          "runtime": {
            "command": "CONTROLLER_ASSIGNMENT",
            "commandTypeId": 1100
          },
          "content": "SELECT '10' AS value UNION ALL SELECT '20';"
        },
        "trigger": {
          "type": "Scheduler",
          "cron": "00 00 00 * * ?",
          "startTime": "1970-01-01 00:00:00",
          "endTime": "9999-01-01 00:00:00",
          "timezone": "Asia/Shanghai",
          "delaySeconds": 0
        },
        "runtimeResource": {
          "resourceGroup": "<RESOURCE_GROUP_IDENTIFIER>"
        },
        "name": "assignment_node_demo",
        "owner": "<OWNER_ID>"
      }
    ]
  }
}

Parameter Spec dari CreateNode berupa string. Jika Anda menggunakan Python SDK, panggil json.dumps(flow_spec, ensure_ascii=False) untuk menserialisasi FlowSpec sebelum meneruskan nilainya ke CreateNodeRequest.spec. Jangan meneruskan langsung dictionary Python.

Tentukan identifier kelompok sumber daya di spec.nodes[].runtimeResource.resourceGroup. Gunakan identifier kelompok sumber daya, bukan nama tampilannya.

Id yang dikembalikan oleh CreateNode adalah ID node aktual yang dihasilkan server dan mungkin berbeda dari nodes[].id dalam FlowSpec. Setelah node dibuat, panggil GetNode dengan Id yang dikembalikan. Verifikasi tipe node, alur kerja, sumber data, dan kelompok sumber daya dalam respons.

Contoh 1: Teruskan hasil kueri MaxCompute SQL

Contoh 2: Teruskan hasil output Python 2

Output pernyataan print Python 2 dipisahkan berdasarkan koma (,) dan diteruskan ke node hilir sebagai array satu dimensi.

• Konfigurasi node hulu (node assignment - Python 2)

  Kode Python 2 sebagai berikut:

  print 'Electronics, Clothing, Books';

• Konfigurasi dan output node hilir (node Shell)

  Di node Shell, tambahkan parameter input bernama types dan referensikan parameter outputs dari node assignment hulu.

  Tulis kode berikut untuk membaca data: # Directly output the entire one-dimensional array echo "Entire result set: ${types}" # Output elements by index echo "Second element: ${types[1]}"

  DataWorks langsung mengurai parameter dan melakukan substitusi statis. Output-nya sebagai berikut: Entire result set: Electronics,Clothing,Books Second element: Clothing

Latar belakang

Asumsikan Anda adalah insinyur pengembangan data di perusahaan internet terpadu, bertanggung jawab untuk memproses data dari tiga lini bisnis inti: e-dagang (ecom), keuangan (finance), dan logistik (logistics), dengan kemungkinan penambahan lini bisnis lain di masa depan. Anda perlu menjalankan logika agregasi yang sama pada log perilaku pengguna dari ketiga lini bisnis ini setiap hari untuk menghitung jumlah tampilan halaman harian (PV) per pengguna dan menyimpan hasilnya di tabel agregat terpadu.

• Tabel sumber hulu (layer DWD):

  • dwd_user_behavior_ecom_d: Tabel perilaku pengguna e-dagang.

  • dwd_user_behavior_finance_d: Tabel perilaku pengguna keuangan.

  • dwd_user_behavior_logistics_d: Tabel perilaku pengguna logistik.

  • dwd_user_behavior_${business_line}_d: Tabel perilaku pengguna untuk lini bisnis potensial lain di masa depan.

  • Tabel-tabel ini memiliki skema yang sama dan dipartisi berdasarkan hari (dt).

• Tabel target hilir (layer DWS):

  • dws_user_summary_d: Tabel agregat pengguna.

  • Tabel ini dipartisi ganda berdasarkan lini bisnis (biz_line) dan hari (dt) untuk menyimpan hasil agregasi dari semua lini bisnis secara terpadu.

Membuat tugas terpisah untuk setiap lini bisnis mengakibatkan biaya maintenance tinggi dan rentan terhadap kesalahan. Dengan node for-each, Anda hanya perlu memelihara satu set logika pemrosesan, dan sistem secara otomatis melakukan iterasi melalui semua lini bisnis untuk menyelesaikan komputasi.

Persiapan data

Pertama, buat tabel sampel dan masukkan data uji (menggunakan tanggal bisnis 20251010 sebagai contoh).

1. Kaitkan sumber daya komputasi dengan workspace.

2. Buka Data Studio untuk pengembangan data dan buat node MaxCompute SQL.

3. Buat tabel sumber (layer DWD): Tambahkan kode berikut ke node MaxCompute SQL dan jalankan.

   -- E-commerce user behavior table
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_ecom_d (
       user_id     STRING COMMENT 'User ID',
       action_type STRING COMMENT 'Action type',
       event_time  BIGINT COMMENT 'Event timestamp in milliseconds (Unix)'
   ) 
   COMMENT 'E-commerce user behavior log detail table'
   PARTITIONED BY (dt STRING COMMENT 'Date partition, format yyyymmdd');
   INSERT OVERWRITE TABLE dwd_user_behavior_ecom_d PARTITION (dt='20251010') VALUES
   ('user001', 'click',        1760004060000), -- 2025-10-10 10:01:00.000
   ('user002', 'browse',       1760004150000), -- 2025-10-10 10:02:30.000
   ('user001', 'add_to_cart',  1760004300000); -- 2025-10-10 10:05:00.000
   -- Verify e-commerce user behavior table created successfully
   SELECT * FROM dwd_user_behavior_ecom_d where dt='20251010';
   -- Finance user behavior table
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_finance_d (
       user_id     STRING COMMENT 'User ID',
       action_type STRING COMMENT 'Action type',
       event_time  BIGINT COMMENT 'Event timestamp in milliseconds (Unix)'
   ) 
   COMMENT 'Finance user behavior log detail table'
   PARTITIONED BY (dt STRING COMMENT 'Date partition, format yyyymmdd');
   INSERT OVERWRITE TABLE dwd_user_behavior_finance_d PARTITION (dt='20251010') VALUES
   ('user003', 'open_app',      1760020200000), -- 2025-10-10 14:30:00.000
   ('user003', 'transfer',      1760020215000), -- 2025-10-10 14:30:15.000
   ('user003', 'check_balance', 1760020245000), -- 2025-10-10 14:30:45.000
   ('user004', 'open_app',      1760020300000); -- 2025-10-10 14:31:40.000
   -- Verify finance user behavior table created successfully
   SELECT * FROM dwd_user_behavior_finance_d where dt='20251010';
   -- Logistics user behavior table
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_logistics_d (
       user_id     STRING COMMENT 'User ID',
       action_type STRING COMMENT 'Action type',
       event_time  BIGINT COMMENT 'Event timestamp in milliseconds (Unix)'
   ) 
   COMMENT 'Logistics user behavior log detail table'
   PARTITIONED BY (dt STRING COMMENT 'Date partition, format yyyymmdd');
   INSERT OVERWRITE TABLE dwd_user_behavior_logistics_d PARTITION (dt='20251010') VALUES
   ('user001', 'check_status',    1760032800000), -- 2025-10-10 18:00:00.000
   ('user005', 'schedule_pickup', 1760032920000); -- 2025-10-10 18:02:00.000
   -- Verify logistics user behavior table created successfully
   SELECT * FROM dwd_user_behavior_logistics_d where dt='20251010';

4. Buat tabel target (layer DWS): Tambahkan kode berikut ke node MaxCompute SQL dan jalankan.

   CREATE TABLE IF NOT EXISTS dws_user_summary_d (
       user_id     STRING COMMENT 'User ID',
       pv          BIGINT COMMENT 'Daily activity count'
   ) 
   COMMENT 'User daily activity summary table'
   PARTITIONED BY (
       dt           STRING COMMENT 'Date partition, format yyyymmdd',
       biz_line     STRING COMMENT 'Business line partition, e.g. ecom, finance, logistics'
   );

   Penting

   Jika workspace menggunakan mode standar, Anda perlu menerapkan node ini ke lingkungan produksi dan melakukan backfill data.

Implementasi alur kerja

1. Buat alur kerja. Di bagian Scheduling Parameters di sisi kanan, atur parameter penjadwalan bizdate ke hari sebelumnya: $[yyyymmdd-1].

2. Di alur kerja, buat node assignment bernama get_biz_list dan tulis kode berikut dalam MaxCompute SQL. Node ini mengeluarkan daftar lini bisnis yang akan diproses:

   -- Output all business lines to be processed
   SELECT 'ecom' AS biz_line
   UNION ALL
   SELECT 'finance' AS biz_line
   UNION ALL
   SELECT 'logistics' AS biz_line;

3. Konfigurasikan node for-each

   • Kembali ke halaman alur kerja dan buat node for-each hilir untuk node assignment get_biz_list.

   • Buka halaman pengaturan node for-each. Di bagian Scheduling Parameters > Script Parameters di bawah schedule settings di sisi kanan, bind parameter loopDataArray ke outputs dari node get_biz_list.

   • Di badan loop node for-each, klik Create Internal Node dan buat node MaxCompute SQL. Tulis logika pemrosesan di dalam badan loop.

     Catatan

     • Skrip ini digerakkan oleh node for-each dan dieksekusi sekali untuk setiap lini bisnis.

     • Variabel bawaan ${dag.foreach.current} secara dinamis diganti dengan nama lini bisnis saat ini pada setiap iterasi. Nilai iterasi yang diharapkan adalah: 'ecom', 'finance', 'logistics'.

     SET odps.sql.allow.dynamic.partition=true;
     INSERT OVERWRITE TABLE dws_user_summary_d PARTITION (dt='${bizdate}', biz_line)
     SELECT
         user_id,
         COUNT(*) AS pv,
         '${dag.foreach.current}' AS biz_line
     FROM
         dwd_user_behavior_${dag.foreach.current}_d
     WHERE
         dt = '${bizdate}'
     GROUP BY
         user_id;

4. Tambahkan node verifikasi

   Kembali ke alur kerja. Klik Create Downstream pada node for-each untuk membuat node MaxCompute SQL dan tambahkan kode berikut.

   SELECT * FROM dws_user_summary_d WHERE dt='20251010' ORDER BY biz_line, user_id;

Penerapan dan hasil

Terapkan alur kerja ke lingkungan produksi. Buka halaman Auto Triggered Task O&M > Auto Triggered Task di Operation Center. Temukan alur kerja target dan lakukan Pengujian asap dengan tanggal bisnis diatur ke '20251010'.

Setelah eksekusi selesai, lihat log waktu proses di instans pengujian. Output yang diharapkan dari node terakhir adalah sebagai berikut:

Keunggulan

• Skalabilitas tinggi: Untuk menambahkan lini bisnis baru, cukup tambahkan satu baris SQL di node assignment tanpa mengubah logika pemrosesan.

• Maintenance mudah: Semua lini bisnis berbagi satu set logika pemrosesan. Satu modifikasi berlaku untuk semuanya.