Realtime Compute for Apache Flink:Flink toplu işleme

Adım 2: DWD ve DWS tabloları oluşturun

Scripts sekmesindeki düzenleyicide aşağıdaki SQL deyimlerini girin ve soldaki Run seçeneğine tıklayın.

USE `my_catalog`.`order_dw`;

CREATE TABLE dwd_orders (
    order_id BIGINT,
    order_user_id STRING,
    order_shop_id BIGINT,
    order_product_id BIGINT,
    order_product_catalog_name STRING,
    order_fee BIGINT,
    order_create_time TIMESTAMP,
    order_update_time TIMESTAMP,
    order_state INT,
    pay_id BIGINT,
    pay_platform INT COMMENT 'platform 0: phone, 1: pc',
    pay_create_time TIMESTAMP
) WITH (
    'sink.parallelism' = '2'
);

CREATE TABLE dws_users (
    user_id STRING,
    ds STRING,
    total_fee BIGINT COMMENT 'Total amount paid on the current day'
) WITH (
    'sink.parallelism' = '2'
);

CREATE TABLE dws_shops (
    shop_id BIGINT,
    ds STRING,
    total_fee BIGINT COMMENT 'Total amount paid on the current day'
) WITH (
    'sink.parallelism' = '2'
);

Adım 3: DWD ve DWS dağıtımlarını oluşturun ve dağıtın

1. DWD dağıtımını oluşturun ve dağıtın.

   1. DWD tablosu güncelleme dağıtımını oluşturun.

      Development > ETL sayfasında dwd_orders adında bir blank batch draft oluşturun. Önerilen etikete sahip bir motor sürümü seçmeniz önerilir. Varsayılan Flink SQL lehçesini kullanın. Aşağıdaki SQL deyimini düzenleyiciye kopyalayın. DWD tablosu bir Paimon ekleme yalnızca (append-only) tablosu olduğundan, tablonun tamamını üzerine yazmak için bir INSERT OVERWRITE deyimi kullanın.

      INSERT OVERWRITE my_catalog.order_dw.dwd_orders
      SELECT 
          o.order_id,
          o.user_id,
          o.shop_id,
          o.product_id,
          c.catalog_name,
          o.buy_fee,
          o.create_time,
          o.update_time,
          o.state,
          p.pay_id,
          p.pay_platform,
          p.create_time
      FROM 
          my_catalog.order_dw.orders as o, 
          my_catalog.order_dw.product_catalog as c, 
          my_catalog.order_dw.orders_pay as p
      WHERE o.product_id = c.product_id AND o.order_id = p.order_id

   2. Sayfanın sağ üst köşesinde Deploy seçeneğine tıklayın ve ardından dwd_orders dağıtımını dağıtmak için OK seçeneğine tıklayın.

2. DWS dağıtımlarını oluşturun ve dağıtın.

   1. DWS tablosu güncelleme dağıtımlarını oluşturun.

      DWD tablosu güncelleme dağıtımını oluşturma adımlarını izleyerek dws_shops ve dws_users adlarında iki toplu iş taslağı oluşturun. Aşağıdaki SQL deyimlerini ilgili düzenleyicilere kopyalayın.

      INSERT OVERWRITE my_catalog.order_dw.dws_shops
      SELECT 
          order_shop_id,
          DATE_FORMAT(pay_create_time, 'yyyyMMdd') as ds,
          SUM(order_fee) as total_fee
      FROM my_catalog.order_dw.dwd_orders
      WHERE pay_id IS NOT NULL AND order_fee IS NOT NULL
      GROUP BY order_shop_id, DATE_FORMAT(pay_create_time, 'yyyyMMdd');

      INSERT OVERWRITE my_catalog.order_dw.dws_users
      SELECT 
          order_user_id,
          DATE_FORMAT(pay_create_time, 'yyyyMMdd') as ds,
          SUM(order_fee) as total_fee
      FROM my_catalog.order_dw.dwd_orders
      WHERE pay_id IS NOT NULL AND order_fee IS NOT NULL
      GROUP BY order_user_id, DATE_FORMAT(pay_create_time, 'yyyyMMdd');

   2. Sayfanın sağ üst köşesinde Deploy seçeneğine tıklayın ve ardından dws_shops ve dws_users dağıtımlarını dağıtmak için OK seçeneğine tıklayın.

Adım 4: DWD ve DWS dağıtımlarını başlatın ve görüntüleyin

• DWD dağıtımını başlatın ve verilerini görüntüleyin.

  1. O&M > Deployments sayfasına gidin. Açılır listeden BATCH seçeneğini belirleyin. dwd_orders dağıtımını bulun ve Actions sütunundaki Start seçeneğine tıklayın.

     Toplu iş örnekleri listesinde STARTING durumunda bir iş örneği görünür.

     İş örneğinin durumu FINISHED olarak değiştiğinde veri işleme tamamlanmış demektir.

  2. Veri sonuçlarını inceleyin.

     Scripts sekmesindeki düzenleyicide aşağıdaki SQL deyimini girin ve DWD tablosundaki verileri sorgulamak için soldaki Run seçeneğine tıklayın.

     SELECT * FROM `my_catalog`.`order_dw`.`dwd_orders`;

     Sonuç aşağıdaki şekilde gösterilmiştir.

     

• DWS dağıtımlarını başlatın ve verilerini görüntüleyin.

  1. O&M > Deployments sayfasında açılır listeden BATCH seçeneğini belirleyin. dws_shops ve dws_users dağıtımlarını bulun ve her biri için Actions sütunundaki Start seçeneğine tıklayın.

  2. Scripts sekmesindeki düzenleyicide aşağıdaki SQL deyimlerini girin ve DWS tablolarındaki verileri sorgulamak için soldaki Run seçeneğine tıklayın.

     SELECT * FROM `my_catalog`.`order_dw`.`dws_shops`;
     SELECT * FROM `my_catalog`.`order_dw`.`dws_users`;

     Sonuçlar aşağıdaki şekillerde gösterilmiştir.

     

Adım 5: Görev orkestrasyonu ile bir toplu iş hattı oluşturun

Bu adımda dağıtımları bir iş akışında orkestre edeceksiniz. Bu sayede dağıtımlar birlikte tetiklenebilir ve sırayla çalıştırılabilir.

1. Bir iş akışı oluşturun.

   1. Sol gezinme bölmesinde O&M > Workflows seçeneğini belirleyin ve ardından Create Workflow seçeneğine tıklayın.

   2. Görüntülenen panelde ad olarak wf_orders girin. Zamanlama türünü varsayılan (Manual Scheduling) olarak bırakın, Resource Queue için default-queue seçeneğini belirleyin ve iş akışı düzenleyicisini açmak için Create seçeneğine tıklayın.

   3. İş akışını düzenleyin.

      1. Başlangıç düğümüne tıklayın, adını v_dwd_orders olarak değiştirin ve bunun için dwd_orders dağıtımını seçin.

      2. v_dws_shops adlı bir düğüm oluşturmak için Add Task seçeneğine tıklayın. Bunun için dws_shops dağıtımını seçin ve v_dwd_orders öğesini yukarı akış düğümü olarak ayarlayın.

      3. v_dws_users adlı bir düğüm oluşturmak için tekrar Add Task seçeneğine tıklayın. Bunun için dws_users dağıtımını seçin ve v_dwd_orders öğesini yukarı akış düğümü olarak ayarlayın.

      4. Sağ üst köşede Save ve ardından OK seçeneğine tıklayın.

2. İş akışını manuel olarak tetikleyin.

   Not

   İş akışını periyodik bir zamanlamaya göre çalışacak şekilde de değiştirebilirsiniz. Workflows sayfasında iş akışının sağındaki Edit Workflow seçeneğine tıklayın ve zamanlama modunu Periodic Scheduling olarak değiştirin. Daha fazla bilgi için bkz. Görev orkestrasyonu (Genel Önizleme).

   1. İş akışını tetiklemeden önce iş akışının yürütme sonuçlarını doğrulamak için ODS tablolarına yeni veriler ekleyin.

      Scripts sekmesindeki düzenleyicide aşağıdaki SQL deyimlerini girin ve soldaki Run seçeneğine tıklayın.

      USE `my_catalog`.`order_dw`;
      
      INSERT INTO orders VALUES
      (100008, 'user_001', 12346, 1, 10000, TO_TIMESTAMP('2023-02-15 17:40:56'), TO_TIMESTAMP('2023-02-15 18:42:56'), 1),
      (100009, 'user_002', 12347, 2, 20000, TO_TIMESTAMP('2023-02-15 18:40:56'), TO_TIMESTAMP('2023-02-15 18:42:56'), 1),
      (100010, 'user_003', 12348, 3, 30000, TO_TIMESTAMP('2023-02-15 19:40:56'), TO_TIMESTAMP('2023-02-15 18:42:56'), 1);
      
      INSERT INTO orders_pay VALUES
      (2008, 100008, 1, TO_TIMESTAMP('2023-02-15 20:40:56')),
      (2009, 100009, 1, TO_TIMESTAMP('2023-02-15 20:40:56')),
      (2010, 100010, 1, TO_TIMESTAMP('2023-02-15 20:40:56'));

      İş durumunu izlemek için Results sekmesinin solundaki Flink UI seçeneğine tıklayın.

   2. O&M > Workflows sayfasında önceki adımda oluşturulan iş akışını bulun, Actions sütunundaki Execute seçeneğine tıklayın ve iş akışını tetiklemek için OK seçeneğine tıklayın.

      İş akışı örnek listesini görüntüleyebileceğiniz Overview sayfasına gitmek için iş akışı adına tıklayın.

      Çalışan iş akışı örneğinin ID değerine tıklayarak yürütme ayrıntıları sayfasına gidin ve her düğümün durumunu gözlemleyin. Tüm iş akışının tamamlanmasını bekleyin.

3. İş akışı yürütme sonuçlarını görüntüleyin.

   1. Scripts sekmesindeki düzenleyicide aşağıdaki SQL deyimlerini girin ve soldaki Run seçeneğine tıklayın.

      SELECT * FROM `my_catalog`.`order_dw`.`dws_shops`;
      SELECT * FROM `my_catalog`.`order_dw`.`dws_users`;

   2. İş akışının yürütme sonuçlarını görüntüleyin.

      İş akışının ODS katmanından gelen yeni verileri işlediğini ve bunları DWS tablolarına yazdığını görebilirsiniz.