DataWorks:Node penetapan

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 adalah sebagai berikut:

  print 'Electronics, Clothing, Books';

• Konfigurasi dan output node hilir (node Shell)

  Di node Shell, tambahkan parameter input bernama types yang mereferensikan outputs dari node assignment hulu.

  Tulis kode berikut untuk membaca data: # Print the entire one-dimensional array echo "Full result set: ${types}" # Print an element from the array by index echo "Second element: ${types[1]}"

  DataWorks langsung mengurai parameter dan melakukan penggantian statis. Output waktu proses adalah sebagai berikut: Full result set: Electronics,Clothing,Books Second element: Clothing

Latar belakang

Asumsikan Anda adalah insinyur data di perusahaan internet besar yang bertanggung jawab memproses data dari tiga lini bisnis inti: e-dagang (ecom), keuangan (finance), dan logistik (logistics). Lebih banyak lini bisnis mungkin ditambahkan di masa depan. Anda perlu menjalankan logika agregasi harian yang sama pada log perilaku pengguna untuk lini bisnis ini guna menghitung PV (page view) untuk setiap pengguna dan menyimpan hasilnya di tabel ringkasan terpadu.

• Tabel sumber hulu (lapisan 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_${biz_line}_d: Tabel perilaku pengguna untuk lini bisnis potensial lainnya.

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

• Tabel target hilir (lapisan DWS):

  • dws_user_summary_d: Tabel ringkasan pengguna.

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

Membuat tugas terpisah untuk setiap lini bisnis sulit dipelihara dan rentan kesalahan. Menggunakan node for-each menyederhanakan pemeliharaan karena Anda hanya perlu memelihara satu set logika pemrosesan, dan sistem secara otomatis melakukan iterasi melalui semua lini bisnis untuk menyelesaikan perhitungan.

Persiapan data

Pertama, buat tabel sampel dan masukkan data uji. Contoh ini menggunakan tanggal bisnis 20251010.

1. Bind compute engine MaxCompute ke workspace.

2. Buka Data Studio dan buat node MaxCompute SQL.

3. Buat tabel sumber (lapisan DWD): Tambahkan kode berikut ke node MaxCompute SQL, pilih, lalu jalankan.

   -- Tabel perilaku pengguna e-dagang
   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 'UNIX timestamp of the event in milliseconds'
   ) 
   COMMENT 'Detailed e-commerce user behavior log 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
   -- Verifikasi bahwa tabel perilaku pengguna e-dagang 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 'Action type',
       event_time  BIGINT COMMENT 'UNIX timestamp of the event in milliseconds'
   ) 
   COMMENT 'Detailed finance user behavior log 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
   -- 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 'Action type',
       event_time  BIGINT COMMENT 'UNIX timestamp of the event in milliseconds'
   ) 
   COMMENT 'Detailed logistics user behavior log 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
   -- Verifikasi bahwa tabel perilaku pengguna logistik telah dibuat.
   SELECT * FROM dwd_user_behavior_logistics_d where dt='20251010';

4. Buat tabel target (lapisan DWS): 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 'PV'
   ) 
   COMMENT 'User daily PV summary table'
   PARTITIONED BY (
       dt           STRING COMMENT 'Date partition, format yyyymmdd',
       biz_line     STRING COMMENT 'Business line partition, such as ecom, finance, logistics'
   );

   Penting

   Jika workspace berada di Standard Edition, Anda harus deploy node ini ke lingkungan produksi dan lakukan pengisian ulang data.

Implementasi workflow

1. Buat workflow. Di panel Scheduling Parameters di sebelah kanan, atur scheduling parameter bizdate ke hari sebelumnya: $[yyyymmdd-1].

2. Di workflow, buat node assignment bernama get_biz_list dan gunakan bahasa MaxCompute SQL untuk menulis kode berikut. 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 kanvas workflow dan buat node for-each hilir untuk node assignment get_biz_list.

   • Buka halaman pengaturan node for-each. Di panel scheduling configuration di sebelah kanan, di bawah Scheduling Parameters > Script Parameters, bind parameter loopDataArray ke outputs dari node get_biz_list.

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

     Catatan

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

     • Saat waktu proses, variabel bawaan ${dag.foreach.current} diganti secara dinamis dengan nama lini bisnis saat ini. 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 workflow. Klik Create Downstream Node untuk node for-each guna membuat node MaxCompute SQL dan tambahkan kode berikut.

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

Penyebaran dan hasil

Deploy workflow ke lingkungan produksi. Buka Auto Triggered Task O&M > Auto Triggered Task, temukan workflow target, lalu jalankan smoke test dengan tanggal bisnis diatur ke '20251010'.

Setelah eksekusi selesai, lihat log eksekusi di instans pengujian. Node terakhir harus menghasilkan output berikut:

Keunggulan

• Skalabilitas tinggi: Untuk menambahkan lini bisnis baru, Anda hanya perlu menambahkan satu baris SQL ke node assignment. Logika pemrosesan tidak perlu diubah.

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