Realtime Compute for Apache Flink：Bangun gudang data waktu nyata dengan Flink dan Hologres

Buat instance ApsaraDB RDS for MySQL dan siapkan data

1. Buat Instance ApsaraDB RDS for MySQL.

   Instance ApsaraDB RDS for MySQL Anda, ruang kerja Flink, dan instance Hologres harus berada di VPC yang sama.

2. Buat Database bernama order_dw dan Akun Standar.

3. Siapkan Sumber Data CDC MySQL.

   1. Di pojok kanan atas halaman detail instance MySQL, klik Log On to Database.

   2. Di dialog, masukkan akun dan kata sandi Anda dan klik Login.

   3. Klik dua kali database order_dw di panel navigasi sisi kiri.

   4. Di tab Konsol SQL, tulis pernyataan DDL untuk membuat tiga tabel bisnis dan sisipkan data ke dalam tabel.

   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 numeric(20,2) 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.05, '2023-02-15 16:40:56', '2023-02-15 18:42:56', 1),
   (100002, 'user_002', 12346, 2, 4000.04, '2023-02-15 15:40:56', '2023-02-15 18:42:56', 1),
   (100003, 'user_003', 12347, 3, 3000.03, '2023-02-15 14:40:56', '2023-02-15 18:42:56', 1),
   (100004, 'user_001', 12347, 4, 2000.02, '2023-02-15 13:40:56', '2023-02-15 18:42:56', 1),
   (100005, 'user_002', 12348, 5, 1000.01, '2023-02-15 12:40:56', '2023-02-15 18:42:56', 1),
   (100006, 'user_001', 12348, 1, 1000.01, '2023-02-15 11:40:56', '2023-02-15 18:42:56', 1),
   (100007, 'user_003', 12347, 4, 2000.02, '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');

4. Klik Execute(F8). Di panel yang muncul, klik Execute.

Buat instance Hologres dan gudang virtual

1. Buat Instance Eksklusif Hologres.

   1. Untuk Product Type, pilih Exclusive instance (subscription) atau Exclusive instance (pay-as-you-go) sesuai kebutuhan.

   2. Untuk Specifications, pilih Compute Group Type.

   3. Untuk Reserved Computing Resources of Virtual Warehouse, masukkan 64 untuk mendukung penambahan gudang virtual tambahan.

   4. Untuk VPC, pilih VPC instance MySQL Anda dari daftar drop-down.

2. Masuk ke instance Hologres Anda dan buat database.

   1. Di panel navigasi sisi kiri konsol Hologres, pilih Instances.

   2. Klik nama instance Hologres target.

   3. Di halaman detail instance, klik Connect to Instance.

   4. Di bilah navigasi atas konsol HoloWeb yang muncul, pilih Create Database.

      1. Untuk Database Name, masukkan order_dw.

      2. Untuk Policy, pilih SPM.

      3. Klik OK.

   5. Tetapkan peran admin ke akun Alibaba Cloud Anda, pengguna RAM, atau peran. Untuk informasi lebih lanjut, lihat Kelola Database.

      Catatan

      • Jika Anda tidak menemukan pengguna RAM atau peran Anda di daftar drop-down User, tambahkan pengguna RAM sebagai superuser instance di halaman Security Center > User Management.

      • Ekspansi log biner diaktifkan secara default di Hologres V2.0 atau lebih baru.

3. Buat gudang virtual bernama read_warehouse_1 untuk menangani pertanyaan.

   Gudang virtual awal init_warehouse digunakan untuk penulisan data.

   Karena sumber daya komputasi cadangan secara otomatis dialokasikan ke gudang virtual awal, Anda harus mengurangi sumber daya komputasinya sebelum membuat instance gudang virtual. Lakukan langkah-langkah berikut:

   1. Di bilah navigasi atas konsol HoloWeb, klik Security Center > Compute Group Management. Konfirmasikan nama instance sudah benar.

   2. Temukan gudang virtual awal init_warehouse dan klik Modify Configuration di kolom Actions.

   3. Di dialog, kurangi sumber daya grup komputasi, dan klik OK.

   4. Klik Create Compute Group. Masukkan read_warehouse_1 di bidang Compute Group Name dan klik OK.

Buat ruang kerja Realtime Compute for Apache Flink dan katalog

1. Buat Ruang Kerja.

   Luncurkan ruang kerja Flink Anda di VPC yang sama dengan instance ApsaraDB RDS for MySQL dan instance Hologres Anda.

2. Masuk ke Konsol Realtime Compute for Apache Flink. Temukan ruang kerja target dan klik Console di kolom Actions.

3. Buat Klaster Sesi, yang menyediakan lingkungan eksekusi untuk membuat katalog dan skrip.

4. Buat katalog Hologres melalui Flink SQL.

   1. Di panel navigasi sisi kiri konsol pengembangan Realtime Compute for Apache Flink, pilih Development > Scripts.

   2. Klik New untuk membuat skrip baru, salin kode berikut ke editor, dan ganti nilai placeholder dengan nilai aktual Anda.

   3. Kemudian, pilih kode yang ingin Anda jalankan, dan klik Run.

      Di pojok kanan bawah halaman, Anda dapat melihat lingkungan eksekusi adalah klaster sesi yang Anda buat.

CREATE CATALOG dw WITH (
  'type' = 'hologres',
  'endpoint' = '<ENDPOINT>', 
  'username' = 'BASIC$flinktest',
  'password' = '${secret_values.holosecrect}',
  'dbname' = 'order_dw@init_warehouse', --Hubungkan ke gudang virtual init_warehouse.
  'binlog' = 'true', -- Opsi konektor yang Anda tentukan saat membuat katalog Hologres juga berlaku untuk tabel baru yang dibuat di katalog ini.
  'sdkMode' = 'jdbc', -- Mode yang direkomendasikan.
  'cdcmode' = 'true',
  'connectionpoolname' = 'the_conn_pool',
  'ignoredelete' = 'true',  -- Aktifkan opsi ini untuk mencegah retraksi data saat mengkonsolidasikan tabel.
  'partial-insert.enabled' = 'true', -- Aktifkan pembaruan parsial saat konsolidasi tabel terlibat.
  'mutateType' = 'insertOrUpdate', -- Aktifkan pembaruan parsial saat konsolidasi tabel terlibat.
  'table_property.binlog.level' = 'replica', -- Kirim properti tabel persisten, pencatatan biner diaktifkan dalam kasus ini, saat membuat katalog. Properti ini berlaku untuk semua tabel yang dibuat kemudian di katalog ini.
  'table_property.binlog.ttl' = '259200'
);

Ganti nilai placeholder berikut:

5. Buat katalog MySQL.

   Salin kode berikut di editor SQL, ganti nilai placeholder, pilih kode yang ingin Anda jalankan, lalu klik Run. Di pojok kanan bawah halaman, Anda dapat melihat lingkungan eksekusi adalah klaster sesi yang Anda buat.

   CREATE CATALOG mysqlcatalog WITH(
     'type' = 'mysql',
     'hostname' = '<hostname>',
     'port' = '<port>',
     'username' = '<username>',
     'password' = '${secret_values.mysql_pw}',
     'default-database' = 'order_dw'
   );

   Ganti nilai placeholder berikut dengan nilai aktual Anda:

   Opsi	Deskripsi
   hostname	Alamat IP atau nama host yang digunakan untuk mengakses database ApsaraDB RDS for MySQL Anda. Untuk mendapatkan endpoint internal, ikuti langkah-langkah berikut: Di panel navigasi sisi kiri konsol ApsaraDB RDS, pilih Instance. Klik nama instance target. Di bagian Basic Information halaman yang muncul, klik View Details di sebelah bidang Network Type. Di panel yang muncul, salin nilai endpoint internal.
   port	Nomor port database ApsaraDB RDS for MySQL Anda. Nilai default: 3306.
   username	Nama pengguna yang digunakan untuk mengakses database ApsaraDB RDS for MySQL Anda.
   password	Kata sandi yang digunakan untuk mengakses database ApsaraDB RDS for MySQL Anda. Untuk keamanan yang lebih baik, contoh ini menggunakan variabel mysql_pw sebagai pengganti rahasia AccessKey teks biasa. Untuk informasi lebih lanjut, lihat Kelola variabel.

Membangun lapisan ODS: Ingest data ke Hologres secara real-time dengan Flink

Anda dapat menjalankan CREATE DATABASE AS (CDAS) yang terkait dengan katalog untuk dengan mudah membuat lapisan ODS. Lapisan ODS sering kali tidak digunakan untuk OLAP atau pertanyaan titik. Sebagai gantinya, ia digunakan sebagai driver peristiwa untuk pekerjaan streaming. Oleh karena itu, mengaktifkan pencatatan biner untuk lapisan ini sudah cukup. Pencatatan biner adalah salah satu kemampuan inti Hologres. Konektor Hologres dapat digunakan untuk membaca semua rekaman dan kemudian mengonsumsi data log biner tambahan.

1. Buat pekerjaan sinkronisasi.

   1. Di panel navigasi sisi kiri, pilih Development > ETL. Di halaman yang muncul, buat draft aliran SQL bernama ODS dan salin kode berikut ke editor SQL:

      CREATE DATABASE IF NOT EXISTS dw.order_dw   -- Karena opsi table_property.binlog.level ditentukan saat pembuatan katalog Hologres, pencatatan biner diaktifkan untuk semua tabel yang dibuat menggunakan pernyataan CDAS. 
      AS DATABASE mysqlcatalog.order_dw INCLUDING all tables -- Anda dapat memilih tabel di database upstream yang perlu diambil ke dalam gudang data. 
      /*+ OPTIONS('server-id'='8001-8004') */ ;   -- Tentukan rentang nilai parameter server-id untuk instance CDC MySQL.

      Catatan

      • Dalam contoh ini, data disinkronkan ke skema publik database order_dw secara default. Anda juga dapat menyinkronkan data ke skema tertentu di database Hologres tujuan. Untuk informasi lebih lanjut, lihat bagian "Gunakan katalog Hologres yang Anda buat sebagai katalog penyimpanan tujuan yang digunakan dalam pernyataan CREATE TABLE AS" dari topik Kelola Katalog Hologres. Setelah Anda menentukan skema, format nama tabel berubah saat Anda menggunakan katalog. Untuk informasi lebih lanjut, lihat bagian "Gunakan katalog Hologres" dari topik Kelola Katalog Hologres.

      • Pembaruan skema tabel sumber tidak tercermin di tabel tujuan hingga data di tabel sumber dihapus, dimasukkan, atau diperbarui.

   2. Di pojok kanan atas editor SQL, klik Deploy.

   3. Mulai pekerjaan.

      1. Di panel navigasi sisi kiri, pilih O&M > Deployments.

      2. Di halaman Deployments, temukan penyebaran bernama ODS dan klik Start di kolom Actions.

      3. Di panel Start Job, pilih opsi Initial Mode, lalu klik tombol Start.

2. Muat data ke gudang virtual.

   Buat dan muat order_dw_tg_default grup tabel, yang menyimpan data di database order_dw, ke gudang virtual read_warehouse_1. Dengan cara ini, gudang virtual read_warehouse_1 menangani pertanyaan eksternal sementara gudang virtual init_warehouse digunakan untuk penulisan data.

   Di bilah navigasi atas halaman HoloWeb, pilih SQL Editor. Setelah memastikan Anda menggunakan instance dan database yang benar, jalankan perintah berikut. Untuk informasi lebih lanjut, lihat Buat Instance Gudang Virtual. Setelah berhasil dieksekusi, Anda dapat melihat grup tabel order_dw_tg_default telah dimuat ke gudang virtual read_warehouse_1.

   -- Tampilkan grup tabel di database saat ini.
   SELECT tablegroup_name FROM hologres.hg_table_group_properties GROUP BY tablegroup_name;
   
   -- Muat grup tabel ke gudang virtual.
   CALL hg_table_group_load_to_warehouse ('order_dw.order_dw_tg_default', 'read_warehouse_1', 1);
   
   -- Lihat apakah grup tabel telah dimuat.
   select * from hologres.hg_warehouse_table_groups;

3. Di pojok kanan atas, ubah gudang virtual menjadi read_warehouse_1 untuk pertanyaan dan analisis.

   

4. Jalankan perintah berikut di Editor SQL untuk melihat hasil sinkronisasi.

   --- Query data di tabel orders. 
   SELECT * FROM orders;
   
   --- Query data di tabel orders_pay. 
   SELECT * FROM orders_pay;
   
   --- Query data di tabel product_catalog. 
   SELECT * FROM product_catalog;

   

Membangun lapisan DWD: Konsolidasi tabel

Kemampuan pembaruan kolom spesifik yang didukung oleh konektor Hologres digunakan untuk membangun lapisan DWD. Anda dapat menggunakan pernyataan INSERT untuk melakukan pembaruan parsial yang efisien. Pertanyaan titik berperforma tinggi berdasarkan penyimpanan data berorientasi kolom dan penyimpanan data hibrid baris-kolom Hologres membantu Anda menanyakan data dari tabel dimensi yang berbeda. Hologres menggunakan arsitektur isolasi sumber daya yang kuat, yang mencegah gangguan antara beban kerja tulis, baca, dan analitik.

1. Buat tabel bernama dwd_orders di lapisan DWD di Hologres dengan menggunakan katalog Hologres Flink.

   Di panel navigasi sisi kiri konsol pengembangan Realtime Compute for Apache Flink, pilih Development > Scripts. Di editor SQL, salin kode berikut, pilih kode tersebut, lalu klik Run.

   -- Saat data dari sumber yang berbeda ditulis ke tabel sink tunggal, nilai null mungkin muncul di kolom mana pun dari tabel. Oleh karena itu, pastikan bahwa bidang di tabel lebar bersifat nullable. 
   CREATE TABLE dw.order_dw.dwd_orders (
     order_id bigint not null,
     order_user_id string,
     order_shop_id bigint,
     order_product_id bigint,
     order_product_catalog_name string,
     order_fee numeric(20,2),
     order_create_time timestamp,
     order_update_time timestamp,
     order_state int,
     pay_id bigint,
     pay_platform int comment 'platform 0: telepon, 1: pc', 
     pay_create_time timestamp,
     PRIMARY KEY(order_id) NOT ENFORCED
   );
   
   -- Anda dapat memodifikasi properti tabel fisik Hologres melalui katalog Hologres. 
   ALTER TABLE dw.order_dw.dwd_orders SET (
     'table_property.binlog.ttl' = '604800' -- Ubah periode timeout data log biner menjadi satu minggu. 
   );

2. Konsumsi data log biner tabel orders dan orders_pay di lapisan ODS secara real-time.

   1. Di panel navigasi sisi kiri, pilih Development > ETL.

   2. Di halaman yang muncul, buat draft aliran SQL bernama DWD dan salin kode berikut ke editor SQL.

      BEGIN STATEMENT SET;
      
      INSERT INTO dw.order_dw.dwd_orders 
       (
         order_id,
         order_user_id,
         order_shop_id,
         order_product_id,
         order_fee,
         order_create_time,
         order_update_time,
         order_state,
         order_product_catalog_name
       ) SELECT o.*, dim.catalog_name 
         FROM dw.order_dw.orders as o
         LEFT JOIN dw.order_dw.product_catalog FOR SYSTEM_TIME AS OF proctime() AS dim
         ON o.product_id = dim.product_id;
      
      INSERT INTO dw.order_dw.dwd_orders 
        (pay_id, order_id, pay_platform, pay_create_time)
         SELECT * FROM dw.order_dw.orders_pay;
      
      END;

   3. Klik Deploy untuk membuat penyebaran dari draft.

   4. Pergi ke O&M > Deployments, dan klik Start di kolom Actions penyebaran.

   Pernyataan SQL di atas menggabungkan tabel orders dengan tabel product_catalog dan menulis hasil akhir ke dwd_orders. Dengan cara ini, data dikonsolidasikan dan ditulis ke tabel dwd_orders secara real-time.

3. Lihat data tabel dwd_orders.

   Di konsol HoloWeb, sambungkan ke instance Hologres dan masuk ke database tujuan. Kemudian, jalankan pernyataan berikut di editor SQL:

   SELECT * FROM dwd_orders;

   

Membangun lapisan DWS: Lakukan komputasi real-time

1. Buat tabel agregat bernama dws_users dan dws_shops di Hologres dengan menggunakan katalog Hologres Flink.

   1. Di panel navigasi sisi kiri, pilih Development > Scripts.

   2. Di editor SQL, salin kode berikut, pilih kode tersebut, lalu klik Run.

      -- Buat tabel agregat dimensi pengguna. 
      CREATE TABLE dw.order_dw.dws_users (
        user_id string not null,
        ds string not null,
        paied_buy_fee_sum numeric(20,2) not null comment 'Total jumlah pembayaran yang selesai pada hari itu',
        primary key(user_id,ds) NOT ENFORCED
      );
      
      -- Buat tabel agregat dimensi vendor. 
      CREATE TABLE dw.order_dw.dws_shops (
        shop_id bigint not null,
        ds string not null,
        paied_buy_fee_sum numeric(20,2) not null comment 'Total jumlah pembayaran yang selesai pada hari itu',
        primary key(shop_id,ds) NOT ENFORCED
      );

2. Konsumsi data di tabel dw.order_dw.dwd_orders di lapisan DWD secara real-time, lakukan agregasi dengan Flink, lalu tulis hasilnya ke tabel di lapisan DWS di Hologres.

   1. Di panel navigasi sisi kiri, pilih Development > ETL.

   2. Di halaman yang muncul, buat draft aliran SQL bernama DWS dan salin kode berikut ke editor SQL.

      BEGIN STATEMENT SET;
      
      INSERT INTO dw.order_dw.dws_users
        SELECT 
          order_user_id,
          DATE_FORMAT (pay_create_time, 'yyyyMMdd') as ds,
          SUM (order_fee)
          FROM dw.order_dw.dwd_orders c
          WHERE pay_id IS NOT NULL AND order_fee IS NOT NULL -- Data streaming pesanan dan data streaming pembayaran telah ditulis ke tabel lebar. 
          GROUP BY order_user_id, DATE_FORMAT (pay_create_time, 'yyyyMMdd');
      
      INSERT INTO dw.order_dw.dws_shops
        SELECT 
          order_shop_id,
          DATE_FORMAT (pay_create_time, 'yyyyMMdd') as ds,
          SUM (order_fee)
         FROM dw.order_dw.dwd_orders c
         WHERE pay_id IS NOT NULL AND order_fee IS NOT NULL -- Data streaming pesanan dan data streaming pembayaran telah ditulis ke tabel lebar. 
         GROUP BY order_shop_id, DATE_FORMAT (pay_create_time, 'yyyyMMdd');
      END;

   3. Klik Deploy untuk membuat penyebaran dari draft.

   4. Pergi ke O&M > Deployments, dan klik Start di kolom Actions penyebaran.

3. Lihat hasil agregasi di lapisan DWS. Hasilnya diperbarui secara real-time berdasarkan perubahan data input.

   1. Lihat hasil di konsol Hologres.

      Tabel dws_users

      SELECT * FROM dws_users;

      

      Tabel dws_shops

      SELECT * FROM dws_shops;

      

   2. Di konsol ApsaraDB RDS, sisipkan catatan data ke tabel orders dan orders_pay di database order_dw.

      INSERT INTO orders VALUES
      (100008, 'user_003', 12345, 5, 6000.02, '2023-02-15 09:40:56', '2023-02-15 18:42:56', 1);
      
      INSERT INTO orders_pay VALUES
      (2008, 100008, 1, '2023-02-15 19:40:56');

   3. Di konsol Hologres, lihat data yang diperbarui.

      Tabel dwd_orders

      SELECT * FROM dwd_orders;

      

      Tabel dws_users

      SELECT * FROM dws_users;

      

      Tabel dws_shops

      SELECT * FROM dws_shops;

      

Lakukan profiling data

Karena pencatatan biner diaktifkan, profiling data dapat dilakukan untuk membantu Anda langsung melihat perubahan data. Selain itu, data di setiap lapisan dipertahankan, sehingga Anda dapat melakukan pertanyaan ad-hoc pada hasil perantara atau memeriksa kebenaran hasil perhitungan akhir.

SELECT *
FROM dw.order_dw.dwd_orders /*+ OPTIONS('binlog'='false') */ 
WHERE order_user_id = 'user_001' and order_create_time > '2023-02-15 12:00:00'; -- Profiling data dalam mode batch mendukung dorongan filter untuk meningkatkan efisiensi eksekusi pekerjaan batch.

Kueri berbasis kunci

Tanyakan tabel agregat di lapisan DWS berdasarkan kunci utama. Hingga satu juta RPS didukung.

Kode sampel berikut menunjukkan cara menanyakan jumlah pembayaran pengguna tertentu pada tanggal tertentu di konsol HoloWeb.

-- holo sql
SELECT * FROM dws_users WHERE user_id ='user_001' AND ds = '20230215';

Kueri rincian pesanan

Lakukan analisis OLAP pada tabel konsolidasi di lapisan DWD.

Kode sampel berikut memberikan contoh cara menanyakan detail pesanan pelanggan pada platform pembayaran tertentu pada Februari 2023 di konsol HoloWeb.

-- holo sql
SELECT * FROM dwd_orders
WHERE order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
AND order_user_id = 'user_001'
AND pay_platform = 0
ORDER BY order_create_time LIMIT 100;

Laporan real-time

Tampilkan laporan real-time berdasarkan data tabel konsolidasi di lapisan DWD. Penyimpanan data hibrid baris-kolom dan penyimpanan data berorientasi kolom Hologres menyediakan kemampuan analitik OLAP tinggi. Pertanyaan data dapat direspons dalam hitungan detik.

Kode sampel berikut memberikan contoh cara menanyakan total volume pesanan dan total jumlah pesanan setiap kategori pada Februari 2023 di konsol HoloWeb.

-- holo sql
SELECT
  TO_CHAR(order_create_time, 'YYYYMMDD') AS order_create_date,
  order_product_catalog_name,
  COUNT(*),
  SUM(order_fee)
FROM
  dwd_orders
WHERE
  order_create_time >= '2023-02-01 00:00:00'  and order_create_time < '2023-03-01 00:00:00'
GROUP BY
  order_create_date, order_product_catalog_name
ORDER BY
  order_create_date, order_product_catalog_name;