Realtime Compute for Apache Flink：Memulai dengan pemrosesan batch Realtime Compute for Apache Flink

Langkah 2: Buat tabel di lapisan DWD dan DWS

Di editor skrip tab Scripts, masukkan dan pilih pernyataan SQL berikut, lalu klik Run di sisi kiri editor skrip.

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 jumlah pembayaran yang lengkap pada hari saat ini'
) WITH (
    'sink.parallelism' = '2'
);

CREATE TABLE dws_shops (
    shop_id BIGINT,
    ds STRING,
    total_fee BIGINT COMMENT 'Total jumlah pembayaran yang lengkap pada hari saat ini'
) WITH (
    'sink.parallelism' = '2'
);

Langkah 3: Buat draf di lapisan DWD dan DWS dan sebarkan draf sebagai penyebaran

1. Buat draf di lapisan DWD dan sebarkan draf sebagai penyebaran.

   1. Buat draf pembaruan untuk tabel di lapisan DWD.

      Di halaman Development > ETL, buat draf batch kosong bernama dwd_orders dan salin pernyataan SQL berikut ke editor skrip. Kami merekomendasikan Anda memilih versi VVR dengan label RECOMMENDED. Secara default, pernyataan Flink SQL digunakan dan dialek SQL tidak tersedia untuk mesin versi sebelumnya. Pernyataan INSERT OVERWRITE digunakan untuk menimpa tabel di lapisan DWD karena tabel di lapisan DWD adalah tabel Apache Paimon append-only.

      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 sudut kanan atas halaman Editor SQL, klik Deploy. Lalu, klik OK untuk menyebarkan draf dwd_orders sebagai penyebaran.

2. Buat draf di lapisan DWS dan sebarkan draf sebagai penyebaran.

   1. Buat draf pembaruan untuk tabel di lapisan DWS.

      Buat dua draf batch kosong bernama dws_shops dan dws_users dengan merujuk ke Buat draf pembaruan untuk tabel di lapisan DWD. Lalu, salin salah satu pernyataan SQL berikut ke editor skrip draf terkait:

      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 sudut kanan atas halaman Editor SQL, klik Deploy. Lalu, klik OK untuk menyebarkan draf dws_shops dan dws_users sebagai penyebaran.

Langkah 4: Mulai dan lihat penyebaran di lapisan DWD dan DWS

• Mulai dan lihat penyebaran di lapisan DWD.

  1. Di halaman O&M > Deployments, pilih BATCH dari daftar drop-down jenis penyebaran. Temukan penyebaran dwd_orders dan klik Start di kolom Actions.

     Pekerjaan batch dalam status STARTING dibuat, seperti yang ditunjukkan pada gambar berikut.

     

     Jika status pekerjaan berubah menjadi FINISHED, pemrosesan data selesai.

  2. Lihat hasil profil data.

     Di editor skrip tab Scripts, masukkan dan pilih pernyataan SQL berikut, lalu klik Run di sisi kiri editor skrip untuk menanyakan data di tabel lapisan DWD:

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

     Gambar berikut menunjukkan hasil kueri.

     

• Mulai dan lihat penyebaran di lapisan DWS.

  1. Di halaman O&M > Deployments, pilih BATCH dari daftar drop-down jenis penyebaran. Temukan penyebaran dws_shops dan dws_users dan klik Start di kolom Actions setiap penyebaran.

  2. Di editor skrip tab Scripts, masukkan dan pilih pernyataan SQL berikut, lalu klik Run di sisi kiri editor skrip untuk menanyakan data di tabel lapisan DWS:

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

     Gambar berikut menunjukkan hasil kueri.

     

Langkah 5: Orkestrasi penyebaran sebagai alur kerja pemrosesan batch

Bab ini menjelaskan cara mengorkestrasi penyebaran yang dibuat di bagian sebelumnya sebagai alur kerja, sehingga Anda dapat menjalankan penyebaran secara seragam berdasarkan urutan tertentu.

1. Buat alur kerja.

   1. Di panel navigasi sisi kiri konsol Realtime Compute for Apache Flink, klik O&M > Workflows. Di halaman yang muncul, klik Create Workflow.

   2. Di panel Buat Alur Kerja, masukkan wf_orders di bidang Nama. Pertahankan nilai parameter Jenis Penjadwalan pada pengaturan default (Manual Scheduling). Pilih default-queue untuk parameter Resource Queue, lalu klik Create. Halaman pengeditan alur kerja akan muncul.

   3. Edit alur kerja.

      1. Klik tugas awal, beri nama tugas v_dwd_orders, dan pilih penyebaran dwd_orders untuk tugas tersebut.

      2. Klik Tambah Tugas untuk membuat tugas bernama v_dws_shops. Pilih penyebaran dws_shops untuk tugas tersebut dan konfigurasikan tugas v_dwd_orders sebagai tugas hulu.

      3. Klik Tambah Tugas lagi untuk membuat tugas bernama v_dws_users. Pilih penyebaran dws_users untuk tugas tersebut dan konfigurasikan tugas v_dwd_orders sebagai tugas hulu.

      4. Di halaman pengeditan alur kerja, klik Save di sudut kanan atas, lalu klik OK.

         Gambar berikut menunjukkan alur kerja yang telah Anda buat:

         

2. Jalankan alur kerja secara manual.

   Catatan

   Anda dapat mengubah jenis penjadwalan alur kerja menjadi Penjadwalan Periodik dengan mengklik Edit Alur Kerja di kolom Tindakan alur kerja pada halaman Workflows. Untuk informasi lebih lanjut, lihat Alur Kerja (pratinjau publik).

   1. Sebelum menjalankan alur kerja, sisipkan beberapa data ke dalam tabel di lapisan ODS untuk memverifikasi hasil eksekusi alur kerja.

      Di editor skrip tab Scripts, masukkan pernyataan SQL berikut, pilih pernyataan tersebut, lalu klik Run di sisi kiri editor skrip.

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

      Di sisi kiri tab Hasil, klik Flink UI untuk melihat status penyebaran.

   2. Di halaman O&M > Workflows, temukan alur kerja yang telah Anda buat dan klik Execute di kolom Actions. Di kotak dialog yang muncul, klik OK untuk menjalankan alur kerja.

      

      Klik nama alur kerja untuk masuk ke halaman detail alur kerja. Di tab Overview, lihat daftar instance tugas.

      

      Klik ID instance yang sedang berjalan di alur kerja untuk masuk ke halaman detail eksekusi instance. Di halaman yang muncul, lihat status eksekusi setiap tugas. Tunggu hingga eksekusi alur kerja selesai.

      

3. Lihat hasil eksekusi alur kerja.

   1. Di editor skrip tab Scripts, masukkan pernyataan SQL berikut, pilih pernyataan tersebut, lalu klik Run di sisi kiri editor skrip.

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

   2. Lihat hasil eksekusi alur kerja.

      Data baru di lapisan ODS diproses dan ditulis ke tabel di lapisan DWS.