DataWorks：Node assignment

Contoh 1: Meneruskan hasil kueri MaxCompute SQL

Contoh 2: Meneruskan output Python 2

Output pernyataan print Python 2 dipisahkan berdasarkan koma (,) dan diteruskan sebagai larik satu dimensi.

• Assignment Node

  Kode Python 2-nya adalah sebagai berikut:

  print 'Electronics, Clothing, Books';

• Shell node

  Tambahkan parameter input bernama param yang mereferensikan outputs dari node assignment. Gunakan skrip berikut untuk membaca data:

  Gunakan skrip berikut untuk membaca data: # Output the full array echo "Full result set: ${types}" # Output a specific element by index echo "Second element: ${types[1]}"

  DataWorks melakukan penggantian statis. Output: Full result set: Electronics,Clothing,Books Second element: Clothing

Latar belakang

Asumsikan Anda adalah developer data di perusahaan internet besar. Anda bertanggung jawab memproses data dari tiga lini bisnis inti: e-dagang, keuangan, dan logistik. Lini bisnis tambahan mungkin ditambahkan di masa depan. Anda perlu menjalankan logika agregasi yang sama pada log perilaku pengguna dari ketiga lini bisnis ini setiap hari. Logika tersebut menghitung PV untuk setiap pengguna dan menyimpan hasilnya di tabel agregat terpadu.

• Tabel sumber upstream (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_${line-of-business}_d: Tabel perilaku pengguna untuk lini bisnis potensial lainnya.

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

• Tabel tujuan downstream (layer DWS):

  • dws_user_summary_d: Tabel agregat pengguna.

  • Tabel ini dipartisi berdasarkan lini bisnis (biz_line) dan hari (dt). Digunakan untuk menyimpan hasil agregasi semua lini bisnis.

Membuat tugas terpisah untuk setiap lini bisnis mahal untuk dipelihara dan rentan terhadap kesalahan. Jika Anda menggunakan node for-each, Anda hanya perlu memelihara satu set logika pemrosesan. Sistem secara otomatis melakukan iterasi semua lini bisnis untuk menyelesaikan perhitungan.

Persiapan data

Pertama, buat tabel contoh dan masukkan data uji. Contoh ini menggunakan stempel waktu data 20251010.

1. Kaitkan resource komputasi MaxCompute dengan ruang kerja.

2. Buka Data Studio dan buat node MaxCompute SQL.

3. Buat tabel sumber: Tambahkan kode berikut ke node MaxCompute SQL, pilih, lalu 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 'Behavior type',
       event_time  BIGINT COMMENT 'Millisecond-level UNIX timestamp of the event occurrence'
   ) 
   COMMENT 'Details of e-commerce user behavioral logs'
   PARTITIONED BY (dt STRING COMMENT 'Date partition in yyyymmdd format');
   
   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 that the e-commerce user behavior table is created.
   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 'Behavior type',
       event_time  BIGINT COMMENT 'Millisecond-level UNIX timestamp of the event occurrence'
   ) 
   COMMENT 'Details of finance user behavioral logs'
   PARTITIONED BY (dt STRING COMMENT 'Date partition in yyyymmdd format');
   
   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 that the finance user behavior table is created.
   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 'Behavior type',
       event_time  BIGINT COMMENT 'Millisecond-level UNIX timestamp of the event occurrence'
   ) 
   COMMENT 'Details of logistics user behavioral logs'
   PARTITIONED BY (dt STRING COMMENT 'Date partition in yyyymmdd format');
   
   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 that the logistics user behavior table is created.
   SELECT * FROM dwd_user_behavior_logistics_d where dt='20251010';

4. Buat tabel tujuan: Tambahkan kode berikut ke node MaxCompute SQL, pilih, lalu jalankan.

   CREATE TABLE IF NOT EXISTS dws_user_summary_d (
       user_id     STRING COMMENT 'User ID',
       pv          BIGINT COMMENT 'Daily popularity'
   ) 
   COMMENT 'Daily user popularity aggregate table'
   PARTITIONED BY (
       dt           STRING COMMENT 'Date partition in yyyymmdd format',
       biz_line     STRING COMMENT 'Line-of-business partition, such as ecom, finance, logistics'
   );

   Penting

   Jika ruang kerja menggunakan mode standar, Anda harus menyebar node ini ke lingkungan produksi dan melakukan pengisian ulang data.

Implementasi alur kerja

1. Buat alur kerja. Di panel Scheduling di sebelah kanan, atur parameter penjadwalan bizdate ke hari sebelumnya $[yyyymmdd-1].

   

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

   -- Output all lines of business 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 kanvas alur kerja dan buat node for-each downstream untuk node assignment.

   • Buka halaman pengaturan node for-each. Di tab Scheduling di sebelah kanan, di bawah Scheduling Parameters > Script Parameters, atur nilai parameter loopDataArray ke outputs dari node get_biz_list.

     

   • Di badan loop node for-each, klik Create Internal Node. Buat node MaxCompute SQL dan tulis logika pemrosesan untuk badan loop.

     Catatan

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

     • Variabel bawaan ${dag.foreach.current} secara dinamis diganti dengan nama lini bisnis saat ini pada waktu proses. Nilai iterasi yang diharapkan adalah ecom, finance, dan 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 kanvas alur kerja. Untuk node for-each, klik Create Descendant Node. Buat node MaxCompute SQL dan tambahkan kode berikut.

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

Sebar dan jalankan

Sebarkan alur kerja ke lingkungan produksi. Di Operation Center, navigasi ke Auto Triggered Node O&M > Auto Triggered Nodes, temukan alur kerja target, jalankan pengujian, dan pilih '20251010' sebagai stempel waktu data.

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

Keunggulan

• Ekstensibilitas tinggi: Jika lini bisnis baru ditambahkan, Anda hanya perlu menambahkan satu baris kode SQL di node assignment. Anda tidak perlu mengubah logika pemrosesan.

• Pemeliharaan mudah: Semua lini bisnis berbagi logika pemrosesan yang sama. Perubahan di satu tempat berlaku untuk semuanya.