DataWorks:Node for-each

Variabel bawaan

Di dalam isi loop for-each, gunakan variabel bawaan berikut untuk mengakses status dan data loop:

for(int i=0;i<data.length;i++) {
   print(data[i]);
}

Jika output hulu berupa array dua dimensi (seperti hasil kueri SQL), Anda juga dapat menggunakan sintaks berikut untuk pengambilan nilai yang tepat:

Skenario 1: Memproses array satu dimensi (output Shell/Python)

• keluaran node penetapan: 2025-11-01, 2025-11-02, 2025-11-03

• Jumlah iterasi: 3

• Pada iterasi kedua:

  • Nilai ${dag.foreach.current} adalah 2025-11-02.

  • Nilai ${dag.loopTimes} adalah 2.

Skenario 2: Memproses array dua dimensi (output SQL)

• output assignment node (MaxCompute SQL):

  +-----+----------+
  | id  | city     |
  +-----+----------+
  | 101 | beijing  |
  | 102 | shanghai |
  +-----+----------+

• Jumlah iterasi: 2

• Pada iterasi kedua:

  • Nilai ${dag.foreach.current} adalah 102,shanghai.

  • Nilai ${dag.loopTimes} adalah 2.

  • Nilai ${dag.foreach.current[0]} adalah 102.

  • Nilai ${dag.foreach.current[1]} adalah shanghai.

Latar belakang

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

• Tabel sumber hulu (lapisan DWD):

  • dwd_user_behavior_ecom_d: Tabel perilaku pengguna e-commerce.

  • 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 hilir (lapisan DWS):

  • dws_user_summary_d: Tabel agregat pengguna.

  • Tabel ini dipartisi berdasarkan lini bisnis (biz_line) dan hari (dt). Tabel ini digunakan untuk menyimpan hasil agregasi dari seluruh lini bisnis.

Membuat tugas terpisah untuk setiap lini bisnis mahal dalam hal maintenance dan rentan terhadap error. Jika Anda menggunakan for-each node, Anda hanya perlu memelihara satu set logika pemrosesan. Sistem secara otomatis melakukan traversal terhadap 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.

   -- Tabel perilaku pengguna e-commerce
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_ecom_d (
       user_id     STRING COMMENT 'User ID',
       action_type STRING COMMENT 'Jenis perilaku',
       event_time  BIGINT COMMENT 'Stempel waktu UNIX tingkat milidetik saat event terjadi'
   ) 
   COMMENT 'Detail log perilaku pengguna e-commerce'
   PARTITIONED BY (dt STRING COMMENT 'Partisi tanggal dalam 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
   -- Verifikasi bahwa tabel perilaku pengguna e-commerce telah dibuat.
   SELECT * FROM dwd_user_behavior_ecom_d where dt='20251010';
   
   -- Tabel perilaku pengguna keuangan
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_finance_d (
       user_id     STRING COMMENT 'User ID',
       action_type STRING COMMENT 'Jenis perilaku',
       event_time  BIGINT COMMENT 'Stempel waktu UNIX tingkat milidetik saat event terjadi'
   ) 
   COMMENT 'Detail log perilaku pengguna keuangan'
   PARTITIONED BY (dt STRING COMMENT 'Partisi tanggal dalam 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
   -- Verifikasi bahwa tabel perilaku pengguna keuangan telah dibuat.
   SELECT * FROM dwd_user_behavior_finance_d where dt='20251010';
   
   -- Tabel perilaku pengguna logistik
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_logistics_d (
       user_id     STRING COMMENT 'User ID',
       action_type STRING COMMENT 'Jenis perilaku',
       event_time  BIGINT COMMENT 'Stempel waktu UNIX tingkat milidetik saat event terjadi'
   ) 
   COMMENT 'Detail log perilaku pengguna logistik'
   PARTITIONED BY (dt STRING COMMENT 'Partisi tanggal dalam 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
   
   -- Verifikasi bahwa tabel perilaku pengguna logistik telah dibuat.
   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 'Popularitas harian'
   ) 
   COMMENT 'Tabel agregat popularitas pengguna harian'
   PARTITIONED BY (
       dt           STRING COMMENT 'Partisi tanggal dalam format yyyymmdd',
       biz_line     STRING COMMENT 'Partisi lini bisnis, seperti ecom, finance, logistics'
   );

   Penting

   Jika ruang kerja menggunakan mode standar, Anda harus mendeploy 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 dalam alur kerja, buat assignment node bernama get_biz_list. Tulis kode berikut dalam MaxCompute SQL. Node ini menghasilkan daftar lini bisnis yang akan diproses.

   -- Output semua lini bisnis yang akan diproses
   SELECT 'ecom' AS biz_line
   UNION ALL
   SELECT 'finance' AS biz_line
   UNION ALL
   SELECT 'logistics' AS biz_line;

3. Konfigurasi for-each node

   • Kembali ke kanvas alur kerja dan buat for-each node hilir untuk assignment node tersebut.

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

     

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

     Catatan

     • Skrip ini dikendalikan oleh for-each node 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 for-each node, 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;

Deploy dan jalankan

Deploy alur kerja ke lingkungan produksi. Di Pusat Operasi, navigasikan 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 assignment node. Anda tidak perlu mengubah logika pemrosesan.

• Maintenance mudah: Semua lini bisnis menggunakan logika pemrosesan yang sama. Perubahan di satu tempat langsung berlaku untuk semuanya.