Realtime Compute for Apache Flink:Pemrosesan batch Flink

Langkah 2: Buat tabel DWD dan DWS

Pada editor di tab Scripts, masukkan pernyataan SQL berikut dan klik Run di sebelah kiri.

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

Langkah 3: Buat dan terapkan deployment DWD dan DWS

1. Buat dan terapkan deployment DWD.

   1. Buat deployment pembaruan tabel DWD.

      Pada halaman Development > ETL, buat blank batch draft bernama dwd_orders. Kami menyarankan memilih versi engine dengan tag recommended. Gunakan dialek Flink SQL default. Salin pernyataan SQL berikut ke dalam editor. Karena tabel DWD adalah tabel append-only Paimon, gunakan pernyataan INSERT OVERWRITE untuk menimpa seluruh tabel.

      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. Di pojok kanan atas halaman, klik Deploy, lalu klik OK untuk menerapkan deployment dwd_orders.

2. Buat dan terapkan deployment DWS.

   1. Buat deployment pembaruan tabel DWS.

      Mengikuti langkah-langkah yang sama seperti pada membuat deployment pembaruan tabel DWD, buat dua draft batch bernama dws_shops dan dws_users. Salin pernyataan SQL berikut ke dalam editor masing-masing.

      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. Di pojok kanan atas halaman, klik Deploy, lalu klik OK untuk menerapkan deployment dws_shops dan dws_users.

Langkah 4: Mulai dan lihat deployment DWD dan DWS

• Mulai deployment DWD dan lihat datanya.

  1. Buka halaman O&M > Deployments. Pilih BATCH dari daftar drop-down. Temukan deployment dwd_orders dan klik Start di kolom Actions.

     Instans pekerjaan dengan status STARTING muncul dalam daftar instans pekerjaan batch.

     Saat status instans pekerjaan berubah menjadi FINISHED, pemrosesan data selesai.

  2. Jelajahi hasil data.

     Pada editor di tab Scripts, masukkan pernyataan SQL berikut dan klik Run di sebelah kiri untuk mengkueri data di tabel DWD.

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

     Hasilnya ditunjukkan pada gambar berikut.

     

• Mulai deployment DWS dan lihat datanya.

  1. Pada halaman O&M > Deployments, pilih BATCH dari daftar drop-down. Temukan deployment dws_shops dan dws_users, lalu klik Start di kolom Actions untuk masing-masing.

  2. Pada editor di tab Scripts, masukkan pernyataan SQL berikut dan klik Run di sebelah kiri untuk mengkueri data di tabel DWS.

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

     Hasilnya ditunjukkan pada gambar-gambar berikut.

     

Langkah 5: Bangun pipeline batch dengan orkestrasi tugas

Pada langkah ini, Anda akan mengatur deployment-deployment tersebut ke dalam alur kerja, sehingga dapat dipicu bersama dan dijalankan secara berurutan.

1. Buat alur kerja.

   1. Di panel navigasi sebelah kiri, pilih O&M > Workflows, lalu klik Create Workflow.

   2. Pada panel yang muncul, masukkan wf_orders sebagai nama. Biarkan jenis penjadwalan sebagai default (Manual Scheduling), pilih default-queue untuk Resource Queue, lalu klik Create untuk membuka editor alur kerja.

   3. Edit alur kerja.

      1. Klik node awal, beri nama v_dwd_orders, dan pilih deployment dwd_orders untuknya.

      2. Klik Add Task untuk membuat node bernama v_dws_shops. Pilih deployment dws_shops untuknya dan tetapkan v_dwd_orders sebagai node hulu.

      3. Klik Add Task lagi untuk membuat node bernama v_dws_users. Pilih deployment dws_users untuknya dan tetapkan v_dwd_orders sebagai node hulu.

      4. Di pojok kanan atas, klik Save lalu OK.

2. Picu alur kerja secara manual.

   Catatan

   Anda juga dapat mengubah alur kerja agar berjalan sesuai jadwal berkala. Pada halaman Workflows, klik Edit Workflow di sebelah kanan alur kerja dan ubah mode penjadwalan menjadi Periodic Scheduling. Untuk informasi lebih lanjut, lihat Orkestrasi tugas (Pratinjau Publik).

   1. Sebelum memicu alur kerja, masukkan data baru ke dalam tabel ODS untuk memverifikasi hasil eksekusi alur kerja.

      Pada editor di tab Scripts, masukkan pernyataan SQL berikut dan klik Run di sebelah kiri.

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

      Klik Flink UI di sebelah kiri tab Results untuk memantau status pekerjaan.

   2. Pada halaman O&M > Workflows, temukan alur kerja yang dibuat pada langkah sebelumnya, klik Execute di kolom Actions, lalu klik OK untuk memicu alur kerja.

      Klik nama alur kerja untuk membuka halaman Workflow Instance List and Details, tempat Anda dapat melihat daftar instans alur kerja.

      Klik ID instans alur kerja yang sedang berjalan untuk membuka halaman detail eksekusinya dan amati status setiap node. Tunggu hingga seluruh alur kerja selesai.

3. Lihat hasil eksekusi alur kerja.

   1. Pada editor di tab Scripts, masukkan pernyataan SQL berikut dan klik Run di sebelah kiri.

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

   2. Lihat hasil eksekusi alur kerja.

      Anda dapat melihat bahwa alur kerja telah memproses data baru dari lapisan ODS dan menuliskannya ke tabel DWS.