DataWorks:Atama düğümü

Konsolda atama düğümü oluşturma ve kullanma

Aşağıdaki örnek, atama düğümünün sonucunu bir Shell düğümüne aktarma yoluyla genel yordamı gösterir. Uygulamada, alt akıştaki düğüm olarak herhangi bir düğüm türü kullanılabilir.

1. Üst akıştaki atama düğümünü yapılandırın

   Hedef iş akışında bir atama düğümü oluşturun ve düzenleyin. MaxCompute SQL, Python 2 veya Shell seçeneklerinden birini seçin ve sonucu alt akıştaki bir düğüme aktarmak istediğiniz kodu yazın.

   print '10,20,30,40'

2. Alt akıştaki Shell düğümünü yapılandırın

   Bir Shell düğümü oluşturun. Shell düğümünün düzenleme sayfasında üst akıştaki sonucu şu şekilde referans alın:

   1. Sağ taraftaki Scheduling Settings panelinde Node Context Parameters sekmesini seçin.

   2. Input Parameters bölümünde Add parameters seçeneğini tıklayın.

   3. Açılan iletişim kutusunda üst akıştaki düğümün çıkış parametresini önceki adımda yapılandırdığınız atama düğümünün outputs parametresine ayarlayın ve geçerli düğümün giriş parametresi için özel bir Parameter Name belirtin (örneğin param).

      Not

      Yapılandırmadan sonra alt akıştaki düğüm, üst akıştaki atama düğümüne otomatik olarak bağımlılık kurar.

   4. Parametre yapılandırmasından sonra, alt akıştaki Shell düğümünün kodunda üst akıştan gelen değeri ${param} biçimini kullanarak kullanabilirsiniz.

3. Alt akıştaki Data Integration düğümünü yapılandırın

   Uygulamada, bir Data Integration çevrimdışı eşitleme düğümü de atama düğümünün alt akışındaki düğüm olarak kullanılabilir:

   1. Bir Data Integration çevrimdışı eşitleme düğümü oluşturun ve bunu atama düğümünün alt akışındaki düğüm olarak bağlayın.

   2. Çevrimdışı eşitleme düğümünün zamanlama yapılandırma sayfasında Node Context Parameters sekmesini tıklayın.

   3. Giriş parametresi eklemek için Add parameters seçeneğini tıklayın ve değer kaynağını üst akıştaki atama düğümünün outputs parametresine ayarlayın.

   4. Parametreyi düğüm kodunda referans alın: Parametre dize türündeyse değeri çift tırnak içine alın (""). Parametre dize dışı bir türdeyse tırnak eklemeyin.

4. Sonucu doğrulayın

   1. İş akışına dönün ve araç çubuğundaki Deploy seçeneğini tıklayın. Tam dağıtımı seçin.

   2. Operation Center'da Node O&M > Auto Triggered Task O&M > Auto Triggered Task sayfasına gidin ve smoke testing gerçekleştirin.

   3. Test örneğinde nihai sonucun beklentilerinizi karşılayıp karşılamadığını kontrol edin.

OpenAPI kullanarak atama düğümü oluşturma

DataWorks OpenAPI'nin CreateNode işlemine çağrı yaparak da bir atama düğümü oluşturabilirsiniz.

Mevcut bir iş akışında düğüm oluşturmak için iş akışını sorgulamak amacıyla ListWorkflowDefinitions çağrısını yapın. Ardından yanıtta yer alan iş akışı nesnesinin en üst düzey Id değerini ContainerId olarak kullanın. Script.Id değerini kullanmayın.

CreateNode çağrısını yaparken aşağıdaki istek parametrelerini yapılandırın:

• ProjectId: DataWorks workspace kimliği.

• Scene: Bir DataWorks projesinde düğüm oluştururken bu parametreyi DATAWORKS_PROJECT olarak ayarlayın.

• ContainerId: Hedef iş akışının en üst düzey Id değeri. Düğümü mevcut bir iş akışında oluşturmak istiyorsanız bu parametreyi belirtin.

• Spec: Serileştirilmiş FlowSpec JSON dizesi.

Aşağıdaki FlowSpec örneği bir MaxCompute SQL atama düğümü oluşturur. Betik dilinin, veri kaynağının ve betik içeriğinin tutarlı olduğundan emin olun.

{
  "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>"
      }
    ]
  }
}

CreateNode işleminin Spec parametresi bir dizedir. Python SDK kullanıyorsanız değeri CreateNodeRequest.spec alanına geçirmeden önce FlowSpec'i serileştirmek için json.dumps(flow_spec, ensure_ascii=False) çağrısını yapın. Python sözlüğünü doğrudan geçirmeyin.

Kaynak grubu tanımlayıcısını spec.nodes[].runtimeResource.resourceGroup alanında belirtin. Görüntülenen ad yerine kaynak grubu tanımlayıcısını kullanın.

CreateNode tarafından döndürülen Id değeri, sunucu tarafından üretilen gerçek düğüm kimliğidir ve FlowSpec'teki nodes[].id değerinden farklı olabilir. Düğüm oluşturulduktan sonra döndürülen Id değeriyle GetNode çağrısını yapın. Yanıttaki düğüm türünü, iş akışını, veri kaynağını ve kaynak grubunu doğrulayın.

Örnek 1: MaxCompute SQL sorgu sonuçlarını aktarma

Örnek 2: Python 2 çıktı sonuçlarını aktarma

Python 2 print ifadesinin çıktısı virgüllere (,) göre bölünür ve alt akıştaki düğümlere tek boyutlu bir dizi olarak aktarılır.

• Üst akıştaki düğüm (atama düğümü - Python 2) yapılandırması

  Python 2 kodu aşağıdaki gibidir:

  print 'Electronics, Clothing, Books';

• Alt akıştaki düğüm (Shell düğümü) yapılandırması ve çıktısı

  Shell düğümünde types adlı bir giriş parametresi ekleyin ve üst akıştaki atama düğümünün outputs parametresini referans alın.

  Veriyi okumak için aşağıdaki kodu yazın: # Directly output the entire one-dimensional array echo "Entire result set: ${types}" # Output elements by index echo "Second element: ${types[1]}"

  DataWorks parametreyi doğrudan ayrıştırır ve statik değiştirme uygular. Çıktı aşağıdaki gibidir: Entire result set: Electronics,Clothing,Books Second element: Clothing

Arka plan

Kapsamlı bir internet şirketinde veri geliştirme mühendisi olduğunuzu ve e-ticaret (ecom), finans (finance) ile lojistik (logistics) olmak üzere üç temel iş kolundan gelen verileri işlemekten sorumlu olduğunuzu varsayın. Gelecekte daha fazla iş kolunun eklenebilme ihtimali bulunmaktadır. Bu üç iş kolundaki kullanıcı davranışı günlüklerinde her gün aynı toplama mantığını çalıştırarak kullanıcı başına günlük sayfa görüntülemelerini (PV) hesaplamak ve sonuçları tek bir toplam tablosunda saklamak gerekmektedir.

• Üst akış kaynak tabloları (DWD katmanı):

  • dwd_user_behavior_ecom_d: E-ticaret kullanıcı davranış tablosu.

  • dwd_user_behavior_finance_d: Finans kullanıcı davranış tablosu.

  • dwd_user_behavior_logistics_d: Lojistik kullanıcı davranış tablosu.

  • dwd_user_behavior_${business_line}_d: Gelecekte ortaya çıkabilecek diğer iş kolları için kullanıcı davranış tabloları.

  • Bu tablolar aynı şemaya sahiptir ve günlere göre (dt) bölümlenmiştir.

• Alt akış hedef tablosu (DWS katmanı):

  • dws_user_summary_d: Kullanıcı toplam tablosu.

  • Bu tablo, tüm iş kollarından gelen toplama sonuçlarını tek bir yerde saklamak üzere iş koluna (biz_line) ve güne (dt) göre çift bölümlüdür.

Her iş kolu için ayrı bir görev oluşturmak yüksek bakım maliyetine neden olur ve hata yapmaya açıktır. For-each düğümü sayesinde tek bir işleme mantığı kümesini koruyabilir ve sistem tüm iş kolları üzerinde otomatik olarak yineleme yaparak hesaplamayı tamamlar.

Veri hazırlığı

Öncelikle örnek tabloları oluşturun ve test verilerini ekleyin (iş tarihi olarak 20251010 kullanılmıştır).

1. Associate a compute resource seçeneğiyle workspace'e bir hesaplama kaynağı ilişkilendirin.

2. Veri geliştirmesi için Data Studio'ya gidin ve bir MaxCompute SQL düğümü oluşturun.

3. Kaynak tabloları (DWD katmanı) oluşturun: Aşağıdaki kodu MaxCompute SQL düğümüne ekleyin ve çalıştırın.

   -- 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. Hedef tabloyu (DWS katmanı) oluşturun: Aşağıdaki kodu MaxCompute SQL düğümüne ekleyin ve çalıştırın.

   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'
   );

   Önemli

   Workspace standard modu kullanıyorsa bu düğümü üretim ortamına dağıtmanız ve verileri geri doldurmanız gerekir.

İş akışı uygulaması

1. Bir iş akışı oluşturun. Sağ taraftaki Scheduling Parameters bölümünde zamanlama parametresi bizdate değerini önceki güne ayarlayın: $[yyyymmdd-1].

2. İş akışında get_biz_list adlı bir atama düğümü oluşturun ve aşağıdaki kodu MaxCompute SQL olarak yazın. Bu düğüm, işlenecek iş kollarının listesini çıktılar:

   -- 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. For-each düğümünü yapılandırın

   • İş akışı sayfasına dönün ve get_biz_list atama düğümü için alt akışta bir for-each düğümü oluşturun.

   • for-each düğümü ayarları sayfasını açın. Sağ taraftaki schedule settings altında Scheduling Parameters > Script Parameters bölümünde loopDataArray parametresini get_biz_list düğümünün outputs değerine bağlayın.

   • For-each düğümünün döngü gövdesinde Create Internal Node seçeneğini tıklayın ve bir MaxCompute SQL düğümü oluşturun. Döngü gövdesi içinde işleme mantığını yazın.

     Not

     • Bu betik for-each düğümü tarafından yönlendirilir ve her iş kolu için bir kez çalıştırılır.

     • ${dag.foreach.current} yerleşik değişkeni, her yinelemede dinamik olarak geçerli iş kolu adıyla değiştirilir. Beklenen yineleme değerleri şunlardır: '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. Doğrulama düğümü ekleyin

   İş akışına dönün. For-each düğümünde Create Downstream seçeneğini tıklayarak bir MaxCompute SQL düğümü oluşturun ve aşağıdaki kodu ekleyin.

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

Dağıtım ve sonuçlar

İş akışını üretim ortamına dağıtın. Operation Center'da Auto Triggered Task O&M > Auto Triggered Task sayfasına gidin. Hedef iş akışını bulun ve iş tarihini '20251010' olarak ayarlayarak smoke testing gerçekleştirin.

Çalıştırma tamamlandığında test örneğindeki çalışma zamanı günlüğünü görüntüleyin. Son düğümün beklenen çıktısı aşağıdaki gibidir:

Avantajlar

• Yüksek ölçeklenebilirlik: Yeni bir iş kolu eklemek için işleme mantığını değiştirmeden atama düğümüne tek bir SQL satırı eklemeniz yeterlidir.

• Kolay bakım: Tüm iş kolları tek bir işleme mantığı kümesini paylaşır. Tek bir değişiklik hepsine uygulanır.