Realtime Compute for Apache Flink：Membangun danau data terpadu streaming dengan Paimon dan StarRocks

Siapkan sumber data CDC MySQL

Topik ini menggunakan instans ApsaraDB RDS for MySQL sebagai contoh. Anda perlu membuat database bernama order_dw dan mengisi tiga tabel bisnis dengan data.

1. Buat instans ApsaraDB RDS for MySQL.

   Penting

   Instans ApsaraDB RDS for MySQL harus berada di VPC yang sama dengan ruang kerja Flink. Jika tidak berada di VPC yang sama, lihat Bagaimana cara mengakses layanan lain lintas VPC?

2. Buat database dan akun (Usang, dialihkan ke "Langkah 1").

   Buat database bernama order_dw. Buat akun istimewa atau akun standar dengan izin baca dan tulis pada database order_dw.

   Buat tiga tabel dan masukkan data ke dalamnya.

   CREATE TABLE `orders` (
     order_id bigint not null primary key,
     user_id varchar(50) not null,
     shop_id bigint not null,
     product_id bigint not null,
     buy_fee bigint not null,   
     create_time timestamp not null,
     update_time timestamp not null default now(),
     state int not null
   );
   
   CREATE TABLE `orders_pay` (
     pay_id bigint not null primary key,
     order_id bigint not null,
     pay_platform int not null, 
     create_time timestamp not null
   );
   
   CREATE TABLE `product_catalog` (
     product_id bigint not null primary key,
     catalog_name varchar(50) not null
   );
   
   -- Siapkan data.
   INSERT INTO product_catalog VALUES(1, 'phone_aaa'),(2, 'phone_bbb'),(3, 'phone_ccc'),(4, 'phone_ddd'),(5, 'phone_eee');
   
   INSERT INTO orders VALUES
   (100001, 'user_001', 12345, 1, 5000, '2023-02-15 16:40:56', '2023-02-15 18:42:56', 1),
   (100002, 'user_002', 12346, 2, 4000, '2023-02-15 15:40:56', '2023-02-15 18:42:56', 1),
   (100003, 'user_003', 12347, 3, 3000, '2023-02-15 14:40:56', '2023-02-15 18:42:56', 1),
   (100004, 'user_001', 12347, 4, 2000, '2023-02-15 13:40:56', '2023-02-15 18:42:56', 1),
   (100005, 'user_002', 12348, 5, 1000, '2023-02-15 12:40:56', '2023-02-15 18:42:56', 1),
   (100006, 'user_001', 12348, 1, 1000, '2023-02-15 11:40:56', '2023-02-15 18:42:56', 1),
   (100007, 'user_003', 12347, 4, 2000, '2023-02-15 10:40:56', '2023-02-15 18:42:56', 1);
   
   INSERT INTO orders_pay VALUES
   (2001, 100001, 1, '2023-02-15 17:40:56'),
   (2002, 100002, 1, '2023-02-15 17:40:56'),
   (2003, 100003, 0, '2023-02-15 17:40:56'),
   (2004, 100004, 0, '2023-02-15 17:40:56'),
   (2005, 100005, 0, '2023-02-15 18:40:56'),
   (2006, 100006, 0, '2023-02-15 18:40:56'),
   (2007, 100007, 0, '2023-02-15 18:40:56');

Kelola metadata

Buat Katalog Paimon

1. Login ke Konsol Realtime Compute for Apache Flink.

2. Di panel navigasi sebelah kiri, buka halaman Metadata Management dan klik Create Catalog.

3. Di tab Built-in Catalog, klik Apache Paimon, lalu klik Next.

4. Masukkan parameter berikut, pilih DLF sebagai jenis penyimpanan, lalu klik OK.

   Item konfigurasi	Deskripsi	Wajib	Keterangan
   metastore	Jenis metastore.	Ya	Dalam contoh ini, pilih dlf.
   catalog name	Nama katalog data DLF. Penting Jika Anda menggunakan pengguna atau role Resource Access Management (RAM), pastikan Anda memiliki izin baca dan tulis pada data DLF. Untuk informasi lebih lanjut, lihat Manajemen otorisasi data.	Ya	Kami merekomendasikan penggunaan DLF 2.5. Anda tidak perlu memasukkan AccessKey atau informasi lainnya. Anda dapat langsung memilih katalog data DLF yang sudah ada. Untuk membuat katalog data, lihat Katalog data. Setelah Anda membuat katalog data bernama paimoncatalog, pilih dari daftar.

5. Buat database order_dw di katalog data. Database ini digunakan untuk menyinkronkan data dari semua tabel di database order_dw MySQL.

   Di panel navigasi sebelah kiri, pilih Data Exploration > Query Script. Klik New untuk membuat kueri sementara.

   -- Gunakan sumber data paimoncatalog.
   USE CATALOG paimoncatalog;
   -- Buat database order_dw.
   CREATE DATABASE order_dw;

   Pesan Pernyataan berikut telah berhasil dieksekusi! menunjukkan bahwa database telah dibuat.

Untuk informasi lebih lanjut tentang cara menggunakan Katalog Paimon, lihat Kelola Katalog Paimon.

Buat Katalog MySQL

1. Di halaman Metadata Management, klik Create Catalog.

2. Di tab Built-in Catalog, klik MySQL, lalu klik Next.

3. Masukkan parameter berikut dan klik OK untuk membuat katalog MySQL bernama mysqlcatalog.

   Item konfigurasi	Deskripsi	Wajib	Keterangan
   catalog name	Nama katalog.	Ya	Masukkan nama kustom dalam bahasa Inggris. Topik ini menggunakan mysqlcatalog sebagai contoh.
   hostname	Alamat IP atau hostname database MySQL.	Ya	Untuk informasi lebih lanjut, lihat Lihat dan kelola titik akhir serta port instans. Karena instans ApsaraDB RDS for MySQL dan ruang kerja Flink yang sepenuhnya dikelola berada di VPC yang sama, masukkan alamat jaringan pribadi.
   port	Nomor port layanan database MySQL. Nilai default adalah 3306.	Tidak	Untuk informasi lebih lanjut, lihat Lihat dan kelola titik akhir serta port instans.
   default-database	Nama database MySQL default.	Ya	Masukkan nama database yang akan disinkronkan, yaitu order_dw dalam topik ini.
   username	Username untuk layanan database MySQL.	Ya	Ini adalah akun yang dibuat di bagian Siapkan sumber data CDC MySQL.
   password	Password untuk layanan database MySQL.	Ya	Ini adalah password untuk akun yang dibuat di bagian Siapkan sumber data CDC MySQL.

Bangun lapisan ODS: Ingesti data database bisnis secara real-time

Gunakan Flink Change Data Capture (CDC) dan pekerjaan ingesti data YAML untuk menyinkronkan data dari MySQL ke Paimon dalam satu langkah guna membangun lapisan ODS.

1. Buat dan mulai pekerjaan ingesti data YAML.

   1. Di Konsol Realtime Compute for Apache Flink, di halaman Data Studio > Data Ingestion, buat pekerjaan draft YAML kosong bernama ods.

   2. Salin kode berikut ke editor. Pastikan untuk memodifikasi parameter seperti username dan password.

      source:
        type: mysql
        name: MySQL Source
        hostname: rm-bp1e********566g.mysql.rds.aliyuncs.com
        port: 3306
        username: ${secret_values.username}
        password: ${secret_values.password}
        tables: order_dw.\.*  # Gunakan ekspresi reguler untuk membaca semua tabel di database order_dw.
        # (Opsional) Sinkronkan data dari tabel yang baru dibuat selama fase inkremental.
        scan.binlog.newly-added-table.enabled: true
        # (Opsional) Sinkronkan komentar tabel dan field.
        include-comments.enabled: true
        # (Opsional) Utamakan pengiriman chunk tak terbatas untuk mencegah potensi error OutOfMemory TaskManager.
        scan.incremental.snapshot.unbounded-chunk-first.enabled: true
        # (Opsional) Aktifkan filter parsing untuk mempercepat pembacaan.
        scan.only.deserialize.captured.tables.changelog.enabled: true
      
      sink:
        type: paimon
        name: Paimon Sink
        catalog.properties.metastore: rest
        catalog.properties.uri: http://cn-beijing-vpc.dlf.aliyuncs.com
        catalog.properties.warehouse: paimoncatalog
        catalog.properties.token.provider: dlf
        
      pipeline:
        name: MySQL to Paimon Pipeline

      Item konfigurasi	Deskripsi	Wajib	Contoh
      catalog.properties.metastore	Jenis metastore. Nilainya tetap rest.	Ya	rest
      catalog.properties.token.provider	Penyedia token. Nilainya tetap dlf.	Ya	dlf
      catalog.properties.uri	URI untuk mengakses DLF Rest Catalog Server. Formatnya http://[region-id]-vpc.dlf.aliyuncs.com. Untuk informasi lebih lanjut tentang ID Wilayah, lihat Titik akhir.	Ya	http://cn-beijing-vpc.dlf.aliyuncs.com
      catalog.properties.warehouse	Nama Katalog DLF.	Ya	paimoncatalog

      Untuk informasi lebih lanjut tentang cara mengoptimalkan performa penulisan Paimon, lihat Optimasi performa Paimon.

   3. Di pojok kanan atas, klik Deploy.

   4. Di halaman Operation Center > Deployments, temukan pekerjaan ods yang baru saja Anda sebarkan. Di kolom Actions, klik Start. Pilih Stateless Start untuk memulai pekerjaan. Untuk informasi lebih lanjut tentang konfigurasi pemulaan pekerjaan, lihat Mulai pekerjaan.

2. Lihat data dari tiga tabel yang disinkronkan dari MySQL ke Paimon.

   Di tab Query Script halaman Data Development > Data Exploration di Konsol Realtime Compute for Apache Flink, salin kode berikut ke skrip kueri. Pilih potongan kode tersebut dan klik Run di pojok kanan atas.

   SELECT * FROM paimoncatalog.order_dw.orders ORDER BY order_id;

   

Bangun lapisan DWD: Tabel lebar berbasis topik

1. Buat tabel lebar DWD Paimon dwd_orders

   Di tab Query Script halaman Data Development > Data Exploration di Konsol Realtime Compute for Apache Flink, salin kode berikut ke skrip kueri. Pilih potongan kode tersebut dan klik Run di pojok kanan atas.

   CREATE TABLE paimoncatalog.order_dw.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,
       PRIMARY KEY (order_id) NOT ENFORCED
   ) WITH (
       'merge-engine' = 'partial-update', -- Gunakan mesin penggabungan partial-update untuk menghasilkan tabel lebar.
       'changelog-producer' = 'lookup' -- Gunakan produsen changelog lookup untuk menghasilkan changelog dengan latensi rendah.
   );

   Pesan Kueri telah dieksekusi menunjukkan bahwa tabel telah dibuat.

2. Konsumsi changelog dari tabel lapisan ODS orders dan orders_pay secara real-time

   Di Konsol Realtime Compute for Apache Flink, di halaman Data Development > ETL, buat pekerjaan aliran SQL bernama dwd, salin kode berikut ke editor SQL, Deploy pekerjaan tersebut, lalu Start secara stateless.

   Pekerjaan SQL ini melakukan join tabel dimensi antara tabel orders dan tabel product_catalog. Hasil join tersebut, bersama dengan data dari tabel orders_pay, ditulis ke tabel dwd_orders. Mesin penggabungan partial-update Paimon digunakan untuk melebarkan data untuk catatan dengan order_id yang sama di tabel orders dan orders_pay.

   SET 'execution.checkpointing.max-concurrent-checkpoints' = '3';
   SET 'table.exec.sink.upsert-materialize' = 'NONE';
   
   SET 'execution.checkpointing.interval' = '10s';
   SET 'execution.checkpointing.min-pause' = '10s';
   
   -- Paimon saat ini tidak mendukung beberapa pernyataan INSERT ke tabel yang sama dalam satu pekerjaan. Oleh karena itu, gunakan UNION ALL.
   INSERT INTO paimoncatalog.order_dw.dwd_orders 
   SELECT 
       o.order_id,
       o.user_id,
       o.shop_id,
       o.product_id,
       dim.catalog_name,
       o.buy_fee,
       o.create_time,
       o.update_time,
       o.state,
       NULL,
       NULL,
       NULL
   FROM
       paimoncatalog.order_dw.orders o 
       LEFT JOIN paimoncatalog.order_dw.product_catalog FOR SYSTEM_TIME AS OF proctime() AS dim
       ON o.product_id = dim.product_id
   UNION ALL
   SELECT
       order_id,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       NULL,
       pay_id,
       pay_platform,
       create_time
   FROM
       paimoncatalog.order_dw.orders_pay;

3. Lihat data di tabel lebar dwd_orders

   Di tab Query Script halaman Data Development > Data Exploration di Konsol Realtime Compute for Apache Flink, salin kode berikut ke skrip kueri. Pilih potongan kode tersebut dan klik Run di pojok kanan atas.

   SELECT * FROM paimoncatalog.order_dw.dwd_orders ORDER BY order_id;

   

Bangun lapisan DWS: Perhitungan metrik

1. Buat tabel agregasi lapisan DWS dws_users dan dws_shops

   Di tab Query Script halaman Data Development > Data Exploration di Konsol Realtime Compute for Apache Flink, salin kode berikut ke skrip kueri. Pilih potongan kode tersebut dan klik Run di pojok kanan atas.

   -- Tabel metrik agregat berdimensi pengguna.
   CREATE TABLE paimoncatalog.order_dw.dws_users (
       user_id STRING,
       ds STRING,
       paid_buy_fee_sum BIGINT COMMENT 'Total jumlah yang dibayarkan pada hari ini',
       PRIMARY KEY (user_id, ds) NOT ENFORCED
   ) WITH (
       'merge-engine' = 'aggregation', -- Gunakan mesin penggabungan agregasi untuk menghasilkan tabel agregasi.
       'fields.paid_buy_fee_sum.aggregate-function' = 'sum' -- Jumlahkan data di field paid_buy_fee_sum untuk menghasilkan hasil agregat.
       -- Karena tabel dws_users tidak dikonsumsi secara streaming di hilir, Anda tidak perlu menentukan produsen changelog.
   );
   
   -- Tabel metrik agregat berdimensi pedagang.
   CREATE TABLE paimoncatalog.order_dw.dws_shops (
       shop_id BIGINT,
       ds STRING,
       paid_buy_fee_sum BIGINT COMMENT 'Total jumlah yang dibayarkan pada hari ini',
       uv BIGINT COMMENT 'Total jumlah pengguna unik yang melakukan pembelian pada hari ini',
       pv BIGINT COMMENT 'Total jumlah pembelian oleh pengguna pada hari ini',
       PRIMARY KEY (shop_id, ds) NOT ENFORCED
   ) WITH (
       'merge-engine' = 'aggregation', -- Gunakan mesin penggabungan agregasi untuk menghasilkan tabel agregasi.
       'fields.paid_buy_fee_sum.aggregate-function' = 'sum', -- Jumlahkan data di field paid_buy_fee_sum untuk menghasilkan hasil agregat.
       'fields.uv.aggregate-function' = 'sum', -- Jumlahkan data di field uv untuk menghasilkan hasil agregat.
       'fields.pv.aggregate-function' = 'sum' -- Jumlahkan data di field pv untuk menghasilkan hasil agregat.
       -- Karena tabel dws_shops tidak dikonsumsi secara streaming di hilir, Anda tidak perlu menentukan produsen changelog.
   );
   
   -- Untuk menghitung tabel agregasi dari perspektif pengguna dan pedagang secara bersamaan, buat tabel antara dengan primary key komposit pengguna dan pedagang.
   CREATE TABLE paimoncatalog.order_dw.dwm_users_shops (
       user_id STRING,
       shop_id BIGINT,
       ds STRING,
       paid_buy_fee_sum BIGINT COMMENT 'Total jumlah yang dibayarkan pengguna di pedagang pada hari ini',
       pv BIGINT COMMENT 'Jumlah kali pengguna melakukan pembelian di pedagang pada hari ini',
       PRIMARY KEY (user_id, shop_id, ds) NOT ENFORCED
   ) WITH (
       'merge-engine' = 'aggregation', -- Gunakan mesin penggabungan agregasi untuk menghasilkan tabel agregasi.
       'fields.paid_buy_fee_sum.aggregate-function' = 'sum', -- Jumlahkan data di field paid_buy_fee_sum untuk menghasilkan hasil agregat.
       'fields.pv.aggregate-function' = 'sum', -- Jumlahkan data di field pv untuk menghasilkan hasil agregat.
       'changelog-producer' = 'lookup', -- Gunakan produsen changelog lookup untuk menghasilkan changelog dengan latensi rendah.
       -- Tabel antara di lapisan DWM umumnya tidak dikueri langsung oleh aplikasi lapisan atas, sehingga Anda dapat mengoptimalkan performa penulisannya.
       'file.format' = 'avro', -- Format penyimpanan baris Avro memberikan performa penulisan yang lebih efisien.
       'metadata.stats-mode' = 'none' -- Mengabaikan statistik meningkatkan biaya kueri OLAP tetapi meningkatkan performa penulisan. Hal ini tidak memengaruhi pemrosesan aliran kontinu.
   );

   Tanggapan Kueri telah dieksekusi menunjukkan bahwa pembuatan berhasil.

2. Ubah data di tabel dwd_orders di lapisan DWD

   Di tab Data Development > ETL di Konsol Realtime Compute for Apache Flink, buat pekerjaan aliran SQL baru bernama dwm. Salin kode berikut ke editor SQL, deploy pekerjaan tersebut, lalu start secara stateless.

   Pekerjaan SQL ini menulis data dari tabel dwd_orders ke tabel dwm_users_shops. Pekerjaan ini menggunakan mesin penggabungan pra-agregasi Paimon untuk secara otomatis menjumlahkan order_fee guna menghitung total jumlah konsumsi per pengguna per pedagang. Pekerjaan ini juga menjumlahkan nilai 1 untuk menghitung jumlah pembelian per pengguna per pedagang.

   SET 'execution.checkpointing.max-concurrent-checkpoints' = '3';
   SET 'table.exec.sink.upsert-materialize' = 'NONE';
   
   SET 'execution.checkpointing.interval' = '10s';
   SET 'execution.checkpointing.min-pause' = '10s';
   
   INSERT INTO paimoncatalog.order_dw.dwm_users_shops
   SELECT
       order_user_id,
       order_shop_id,
       DATE_FORMAT (pay_create_time, 'yyyyMMdd') as ds,
       order_fee,
       1 -- Satu catatan input mewakili satu pembelian.
   FROM paimoncatalog.order_dw.dwd_orders
   WHERE pay_id IS NOT NULL AND order_fee IS NOT NULL;

3. Konsumsi changelog dari tabel lapisan DWM dwm_users_shops secara real-time

   Di halaman Data Development > ETL di Konsol Realtime Compute for Apache Flink, buat pekerjaan aliran SQL baru bernama dws. Salin kode berikut ke editor SQL, deploy pekerjaan tersebut, lalu start secara stateless.

   Pekerjaan SQL ini menulis data dari tabel dwm_users_shops ke tabel dws_users dan dws_shops. Pekerjaan ini menggunakan mesin penggabungan pra-agregasi Paimon untuk menghitung total jumlah konsumsi untuk setiap pengguna (paid_buy_fee_sum) di tabel dws_users. Di tabel dws_shops, pekerjaan ini menghitung total pendapatan untuk setiap pedagang (paid_buy_fee_sum), jumlah pengguna yang melakukan pembelian (dengan menjumlahkan 1), dan total jumlah pembelian (pv).

   SET 'execution.checkpointing.max-concurrent-checkpoints' = '3';
   SET 'table.exec.sink.upsert-materialize' = 'NONE';
   
   SET 'execution.checkpointing.interval' = '10s';
   SET 'execution.checkpointing.min-pause' = '10s';
   
   -- Berbeda dengan pekerjaan DWD, setiap pernyataan INSERT di sini menulis ke tabel Paimon yang berbeda, sehingga dapat berada dalam satu pekerjaan yang sama.
   BEGIN STATEMENT SET;
   
   INSERT INTO paimoncatalog.order_dw.dws_users
   SELECT 
       user_id,
       ds,
       paid_buy_fee_sum
   FROM paimoncatalog.order_dw.dwm_users_shops;
   
   -- Primary key-nya adalah pedagang. Volume data untuk beberapa pedagang populer mungkin jauh lebih tinggi daripada yang lain.
   -- Oleh karena itu, gunakan local merge untuk pra-agregasi data di memori sebelum menulis ke Paimon. Hal ini membantu mengurangi kesenjangan data.
   INSERT INTO paimoncatalog.order_dw.dws_shops /*+ OPTIONS('local-merge-buffer-size' = '64mb') */
   SELECT
       shop_id,
       ds,
       paid_buy_fee_sum,
       1, -- Satu catatan input mewakili seluruh konsumsi oleh satu pengguna di pedagang tersebut.
       pv
   FROM paimoncatalog.order_dw.dwm_users_shops;
   
   END;

4. Lihat data di tabel dws_users dan dws_shops

   Di tab Query Script halaman Data Development > Data Exploration di Konsol Realtime Compute for Apache Flink, salin kode berikut ke skrip kueri. Pilih potongan kode tersebut dan klik Run di pojok kanan atas.

   --Lihat data di tabel dws_users.
   SELECT * FROM paimoncatalog.order_dw.dws_users ORDER BY user_id;

   

   --Lihat data di tabel dws_shops.
   SELECT * FROM paimoncatalog.order_dw.dws_shops ORDER BY shop_id;

   

Tangkap perubahan di database bisnis

Setelah danau data terpadu aliran dibangun, Anda dapat menguji kemampuannya dalam menangkap perubahan dari database bisnis.

1. Masukkan data berikut ke database order_dw di MySQL.

   INSERT INTO orders VALUES
   (100008, 'user_001', 12345, 3, 3000, '2023-02-15 17:40:56', '2023-02-15 18:42:56', 1),
   (100009, 'user_002', 12348, 4, 1000, '2023-02-15 18:40:56', '2023-02-15 19:42:56', 1),
   (100010, 'user_003', 12348, 2, 2000, '2023-02-15 19:40:56', '2023-02-15 20:42:56', 1);
   
   INSERT INTO orders_pay VALUES
   (2008, 100008, 1, '2023-02-15 18:40:56'),
   (2009, 100009, 1, '2023-02-15 19:40:56'),
   (2010, 100010, 0, '2023-02-15 20:40:56');

2. Lihat data di tabel dws_users dan dws_shops. Di tab Query Script halaman Data Development > Data Exploration di Konsol Realtime Compute for Apache Flink, salin kode berikut ke skrip kueri. Pilih potongan kode tersebut dan klik Run di pojok kanan atas.

   • tabel dws_users

     SELECT * FROM paimoncatalog.order_dw.dws_users ORDER BY user_id;

     

   • tabel dws_shops

     SELECT * FROM paimoncatalog.order_dw.dws_shops ORDER BY shop_id;