Hologres:Bangun gudang data real-time dengan Flink dan Hologres

Buat instans RDS MySQL dan siapkan sumber data

1. Buat instans RDS MySQL. Untuk informasi lebih lanjut, lihat Buat instans RDS MySQL.

   Instans RDS MySQL harus berada dalam VPC yang sama dengan ruang kerja Flink dan instans Hologres Anda.

2. Buat database dan akun.

   Buat database bernama `order_dw` dan akun standar dengan izin baca dan tulis pada database ini. Untuk informasi lebih lanjut, lihat Buat database dan Buat akun.

3. Siapkan sumber data CDC MySQL.

   1. Pada halaman detail instans, klik Log On to Database.

   2. Pada halaman login, masukkan nama akun dan kata sandi database yang telah Anda buat, lalu klik Log on.

   3. Setelah login, klik dua kali database `order_dw` pada halaman instans database untuk beralih ke database tersebut.

   4. Di SQL Console, Anda dapat menulis pernyataan DDL untuk membuat ketiga tabel bisnis dan memasukkan data sampel.

      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
      );
      
      -- Prepare 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 Upload, lalu klik Direct execution.

Buat instans Hologres dan kelompok komputasi

1. Buat instans Hologres dedicated. Untuk informasi lebih lanjut, lihat Beli instans Hologres.

   Instans Hologres harus berada dalam VPC yang sama dengan instans RDS MySQL. Untuk merasakan isolasi sumber daya kuat Hologres menggunakan pemisahan baca/tulis, pilih Virtual Warehouse sebagai tipe instans dan atur Reserved compute resources for compute groups ke 64 untuk mengaktifkan penambahan kelompok komputasi baru.

2. Login ke instans, lalu buat database dan berikan izin.

   Buat database bernama order_dw dan aktifkan model izin sederhana. Kemudian, berikan izin admin kepada pengguna Anda. Untuk informasi lebih lanjut tentang langkah pembuatan dan otorisasi database, lihat Manajemen DB.

   Catatan

   • Jika akun target tidak muncul dalam daftar drop-down User Account, artinya akun tersebut belum ditambahkan ke instans saat ini. Anda dapat membuka halaman Users dan menambahkan pengguna sebagai SuperUser.

   • Hologres V2.0 dan versi setelahnya mengaktifkan ekstensi Binlog secara default. Tidak diperlukan tindakan manual.

3. Tambahkan kelompok komputasi.

   Anda dapat menggunakan kelompok komputasi terpisah untuk isolasi sumber daya. Gunakan kelompok komputasi awal `init_warehouse` untuk ingesti data dan `read_warehouse_1` untuk kueri layanan.

   Semua sumber daya komputasi yang dicadangkan awalnya dialokasikan ke `init_warehouse`. Anda harus mengurangi alokasinya sebelum membuat kelompok komputasi baru. Untuk informasi lebih lanjut, lihat Buat instans kelompok komputasi baru.

   1. Klik Security Center > Virtual Warehouse Management dan pastikan nama instans sesuai dengan instans target Anda.

   2. Klik Modify Configuration di kolom Actions untuk kelompok komputasi `init_warehouse` yang ada. Kurangi sumber dayanya dan klik OK.

   3. Klik Create Virtual Warehouse, beri nama `read_warehouse_1`, lalu klik OK.

Buat ruang kerja Flink dan katalog

1. Buat ruang kerja Flink. Untuk informasi lebih lanjut, lihat Aktifkan Realtime Compute for Apache Flink.

   Ruang kerja Flink harus berada dalam VPC yang sama dengan instans RDS MySQL dan Hologres.

2. Login ke Konsol Realtime Compute dan klik Console di ruang kerja target.

3. Buat kluster session untuk menyediakan lingkungan eksekusi bagi pembuatan katalog dan skrip kueri selanjutnya. Untuk informasi lebih lanjut, lihat Langkah 1: Buat kluster session.

4. Buat katalog Hologres.

   Pada tab Query Script halaman Data Development > Data Query, salin kode berikut ke skrip kueri, perbarui parameter dengan informasi layanan Hologres Anda yang sebenarnya, pilih blok kode tersebut, lalu klik Run. Anda harus menggunakan kluster session yang telah dibuat sebagai lingkungan eksekusi di pojok kanan bawah halaman.

   CREATE CATALOG dw WITH (
     'type' = 'hologres',
     'endpoint' = '<ENDPOINT>', 
     'username' = 'BASIC$flinktest',
     'password' = '${secret_values.holosecrect}',
     'dbname' = 'order_dw@init_warehouse', -- Nama database, menentukan koneksi ke kelompok komputasi init_warehouse.
     'binlog' = 'true', -- Atur parameter WITH default untuk tabel sumber, tabel dimensi, dan tabel sink saat membuat katalog. Parameter ini akan diterapkan otomatis ke tabel di bawah katalog ini.
     'sdkMode' = 'jdbc', -- Gunakan mode JDBC.
     'cdcmode' = 'true',
     'connectionpoolname' = 'the_conn_pool',
     'ignoredelete' = 'true',  -- Diperlukan untuk penggabungan tabel lebar agar mencegah retraksi.
     'partial-insert.enabled' = 'true', -- Diperlukan untuk penggabungan tabel lebar agar mengaktifkan pembaruan kolom parsial.
     'mutateType' = 'insertOrUpdate', -- Diperlukan untuk penggabungan tabel lebar agar mengaktifkan pembaruan kolom parsial.
     'table_property.binlog.level' = 'replica', -- Teruskan properti tabel fisik Hologres yang persisten saat pembuatan katalog. Binlog diaktifkan secara default untuk semua tabel yang dibuat selanjutnya.
     'table_property.binlog.ttl' = '259200'
   );

   Perbarui parameter berikut berdasarkan informasi layanan Hologres Anda yang sebenarnya.

   Parameter	Deskripsi	Catatan
   endpoint	Alamat endpoint Hologres.	Dapatkan nama domain untuk jenis jaringan VPC tertentu dari halaman detail instans Hologres. Untuk detail nama domain, lihat Endpoint.
   username	Pilih salah satu dari berikut: Username akun kustom dalam format BASIC$<user_name>. ID AccessKey akun Alibaba Cloud atau Pengguna RAM Anda.	Pengguna yang dikonfigurasi harus memiliki akses ke database Hologres target. Untuk detail tentang izin database Hologres dan manajemen pengguna, lihat Model izin Hologres dan Manajemen pengguna. Contoh ini menggunakan akun kustom bernama BASIC$flinktest dan mengatur nilai kata sandinya menggunakan variabel proyek bernama holosecrect untuk menghindari risiko keamanan menyimpan kata sandi dalam teks biasa. Untuk informasi lebih lanjut, lihat Variabel Proyek.
   password	Kata sandi akun kustom. Rahasia AccessKey akun Alibaba Cloud atau Pengguna RAM Anda.

   Catatan

   Saat membuat katalog, Anda dapat mengatur parameter `WITH` default untuk tabel sumber, tabel dimensi, dan tabel sink, beserta properti default untuk tabel fisik Hologres (misalnya, parameter yang diawali dengan `table_property`). Untuk informasi lebih lanjut, lihat Kelola katalog Hologres dan Parameter WITH Hologres untuk gudang data real-time.

5. Buat katalog MySQL.

   Salin kode berikut ke Query Script, perbarui parameter dengan informasi layanan MySQL Anda yang sebenarnya, pilih blok kode tersebut, lalu klik Run. Anda harus menggunakan kluster session yang telah dibuat sebagai lingkungan eksekusi di pojok kanan bawah halaman.

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

   Perbarui parameter berikut berdasarkan informasi layanan MySQL Anda yang sebenarnya.

   Parameter	Deskripsi
   hostname	Alamat IP atau hostname database MySQL. Pada halaman informasi dasar database, klik View Connection Details di bagian Network Type untuk mendapatkan alamat jaringan pribadi.
   port	Nomor port layanan database MySQL. Nilai default adalah 3306.
   username	Username untuk layanan database MySQL.
   password	Kata sandi untuk layanan database MySQL. Contoh ini menggunakan variabel bernama mysql_pw untuk menentukan nilai kata sandi, menghindari risiko terkait teks biasa. Untuk detail, lihat Manajemen variabel.

Bangun lapisan ODS: Ingest tabel database operasional secara real-time

Gunakan pernyataan CREATE DATABASE AS (CDAS) dengan katalog untuk membuat seluruh lapisan ODS sekaligus. Lapisan ODS biasanya tidak secara langsung mendukung OLAP atau kueri titik key-value (KV). Sebaliknya, lapisan ini berfungsi sebagai sumber berbasis event untuk pekerjaan streaming. Mengaktifkan Binlog memenuhi persyaratan ini. Binlog merupakan kemampuan inti Hologres, dan konektor Hologres mendukung mode full-and-incremental—pertama membaca data lengkap, lalu mengonsumsi Binlog secara inkremental.

1. Buat pekerjaan sinkronisasi CDAS untuk lapisan ODS.

   1. Pada halaman Data Development > ETL, buat pekerjaan streaming SQL baru bernama `ODS` dan salin kode berikut ke editor SQL.

      CREATE DATABASE IF NOT EXISTS dw.order_dw   -- Parameter table_property.binlog.level telah diatur saat membuat katalog, sehingga semua tabel yang dibuat melalui CDAS memiliki Binlog yang diaktifkan.
      AS DATABASE mysqlcatalog.order_dw INCLUDING all tables -- Pilih tabel dari database hulu yang perlu diingesti.
      /*+ OPTIONS('server-id'='8001-8004') */ ;   -- Tentukan rentang server-id untuk instans CDC MySQL.

      Catatan

      • Secara default, contoh ini menyinkronkan data ke skema Public di database `order_dw`. Anda juga dapat menyinkronkan data ke skema tertentu di database Hologres target. Untuk informasi lebih lanjut, lihat Gunakan sebagai katalog target CDAS. Menentukan skema mengubah format penamaan tabel saat Anda menggunakan katalog. Untuk informasi lebih lanjut, lihat Gunakan katalog Hologres.

      • Jika skema tabel sumber berubah, skema tabel hasil hanya mencerminkan perubahan tersebut setelah modifikasi data—seperti operasi delete, insert, atau update—terjadi di tabel sumber.

   2. Klik Deploy di pojok kanan atas untuk menerapkan pekerjaan.

   3. Di panel navigasi kiri, buka Operation Center > Job Operations. Klik Start di kolom Actions untuk pekerjaan `ODS` yang baru diterapkan. Pilih Stateless start dan klik Start.

2. Muat data ke kelompok komputasi.

   Kelompok tabel digunakan untuk menyimpan data di Hologres. Saat Anda menggunakan `read_warehouse_1` untuk mengkueri data dari kelompok tabel di database `order_dw` (dalam contoh ini, `order_dw_tg_default`. Untuk informasi langkah pembuatan, lihat Manajemen Kelompok Tabel), Anda harus memuat `order_dw_tg_default` untuk kelompok komputasi `read_warehouse_1`. Hal ini memungkinkan Anda menggunakan kelompok komputasi init_warehouse untuk menulis data dan menggunakan kelompok komputasi read_warehouse_1 untuk melakukan pencarian layanan.

   Di halaman pengembangan HoloWeb, klik SQL Editor. Konfirmasi nama instans dan database, lalu jalankan perintah berikut. Untuk informasi selengkapnya, lihat Buat instans kelompok komputasi baru. Setelah data dimuat, verifikasi bahwa `read_warehouse_1` telah memuat data dari kelompok tabel `order_dw_tg_default`.

   -- Lihat kelompok tabel di database saat ini.
   SELECT tablegroup_name FROM hologres.hg_table_group_properties GROUP BY tablegroup_name;
   
   -- Muat kelompok tabel ke kelompok komputasi.
   CALL hg_table_group_load_to_warehouse ('order_dw.order_dw_tg_default', 'read_warehouse_1', 1);
   
   -- Lihat status pemuatan kelompok tabel untuk kelompok komputasi.
   select * from hologres.hg_warehouse_table_groups;

3. Alihkan kelompok komputasi ke `read_warehouse_1` di pojok kanan atas. Anda dapat menggunakan kelompok komputasi ini untuk kueri dan analisis selanjutnya.

   

4. Pada halaman SQL Editor, Anda dapat menjalankan perintah berikut untuk melihat ketiga tabel yang disinkronkan dari MySQL ke Hologres.

   --- Kueri data di orders.
   SELECT * FROM orders;
   
   --- Kueri data di orders_pay.
   SELECT * FROM orders_pay;
   
   --- Kueri data di product_catalog.
   SELECT * FROM product_catalog;

   

Bangun lapisan DWD: Buat tabel lebar real-time

Pembangunan lapisan DWD menggunakan kemampuan pembaruan kolom parsial konektor Hologres, yang memungkinkan pernyataan DML `INSERT` dengan mudah mengekspresikan pembaruan parsial. Pekerjaan ini mengkueri beberapa tabel dimensi menggunakan kemampuan kueri titik berkinerja-tinggi Hologres, yang diaktifkan oleh penyimpanan baris dan penyimpanan hibrida baris/kolom. Selain itu, isolasi sumber daya kuat Hologres memastikan bahwa beban kerja tulis, baca, dan analisis tidak saling mengganggu.

1. Buat tabel lebar DWD `dwd_orders` di Hologres menggunakan fungsi katalog Flink.

   Pada tab Query Script halaman Data Development > Data Query, salin kode berikut ke skrip kueri, pilih blok kode tersebut, lalu klik Run.

   -- Kolom tabel lebar harus nullable karena aliran berbeda menulis ke tabel hasil yang sama, dan kolom apa pun mungkin berisi nilai null.
   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: phone, 1: pc', 
     pay_create_time timestamp,
     PRIMARY KEY(order_id) NOT ENFORCED
   );
   
   -- Ubah properti tabel fisik Hologres melalui katalog.
   ALTER TABLE dw.order_dw.dwd_orders SET (
     'table_property.binlog.ttl' = '604800' -- Atur TTL Binlog menjadi satu minggu.
   );

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

   Pada halaman Data Development > ETL, buat pekerjaan streaming SQL baru bernama `DWD`, salin kode berikut ke editor SQL, lalu deploy dan Start pekerjaan tersebut. Pekerjaan SQL ini menggabungkan tabel `orders` dengan tabel dimensi `product_catalog` dan menulis hasil akhir ke `dwd_orders`. Hal ini mencapai pembangunan tabel lebar real-time.

   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. Lihat data di tabel lebar `dwd_orders`.

   Setelah Anda terhubung ke instans Hologres dan login ke database target pada halaman pengembangan HoloWeb, Anda dapat menjalankan perintah berikut di editor SQL.

   SELECT * FROM dwd_orders;

   

Bangun lapisan DWS: Hitung metrik real-time

1. Buat tabel agregasi DWS `dws_users` dan `dws_shops` di Hologres menggunakan fungsi katalog Flink.

   Pada tab Query Script halaman Data Development > Data Query, salin kode berikut ke skrip kueri, pilih blok kode tersebut, lalu klik Run.

   -- Tabel metrik agregasi berdimensi 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 yang dibayar pada hari tersebut',
     primary key(user_id,ds) NOT ENFORCED
   );
   
   -- Tabel metrik agregasi berdimensi pedagang.
   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 yang dibayar pada hari tersebut',
     primary key(shop_id,ds) NOT ENFORCED
   );

2. Konsumsi tabel lebar DWD `dw.order_dw.dwd_orders` secara real-time, lakukan agregasi di Flink, dan tulis hasilnya ke tabel DWS Hologres.

   Pada halaman Data Development > ETL, buat pekerjaan streaming SQL baru bernama `DWS`, salin kode berikut ke editor SQL, lalu deploy dan Start pekerjaan tersebut.

   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 -- Aliran pesanan dan 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 -- Aliran pesanan dan pembayaran telah ditulis ke tabel lebar.
      GROUP BY order_shop_id, DATE_FORMAT (pay_create_time, 'yyyyMMdd');
   END;

3. Lihat hasil agregasi DWS, yang diperbarui secara real-time seiring perubahan data hulu.

   1. Lihat data di konsol Hologres sebelum perubahan.

      tabel dws_users

      SELECT * FROM dws_users;

      

      tabel dws_shops

      SELECT * FROM dws_shops;

      

   2. Masukkan satu catatan baru ke masing-masing tabel `orders` dan `orders_pay` di database `order_dw` dari konsol RDS.

      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. Lihat data di konsol Hologres setelah perubahan.

      tabel dwd_orders

      SELECT * FROM dwd_orders;

      

      tabel dws_users

      SELECT * FROM dws_users;

      

      tabel dws_shops

      SELECT * FROM dws_shops;

      

Eksplorasi data

Karena Binlog diaktifkan, Anda dapat langsung mengamati perubahan data. Untuk melakukan eksplorasi ad-hoc terhadap hasil antara atau memvalidasi kebenaran komputasi akhir, Anda dapat dengan mudah memeriksa status antara karena setiap lapisan dalam solusi ini menyimpan data secara persisten.

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'; -- Mode batch mendukung filter pushdown untuk meningkatkan efisiensi eksekusi.

Layanan Key-Value

Anda dapat mengkueri tabel metrik agregat pada lapisan DWS berdasarkan primary key, dengan dukungan jutaan RPS.

Contoh berikut menunjukkan cara mengkueri pengeluaran pengguna tertentu pada tanggal tertentu di halaman pengembangan HoloWeb.

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

Kueri detail

Anda dapat melakukan analisis OLAP pada tabel lebar DWD.

Contoh berikut menunjukkan cara mengkueri detail pesanan pelanggan tertentu pada Februari 2023 di platform pembayaran tertentu di halaman pengembangan 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;

Pelaporan real-time

Anda dapat menghasilkan laporan real-time berdasarkan data tabel lebar DWD. Penyimpanan hibrida baris/kolom dan tabel berorientasi kolom Hologres memberikan performa OLAP yang sangat baik dengan waktu respons di bawah satu detik.

Contoh berikut menunjukkan cara mengkueri jumlah total pesanan dan jumlah total pesanan per kategori produk untuk Februari 2023 di halaman pengembangan 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;