Realtime Compute for Apache Flink:Analisis perilaku pengguna dengan Flink, MongoDB, dan Hologres

Arsitektur dan manfaat

Arsitektur

Realtime Compute for Apache Flink adalah kerangka pemrosesan aliran yang kuat yang mendukung pemrosesan efisien sejumlah besar data real-time. ApsaraDB untuk MongoDB adalah database NoSQL berorientasi dokumen yang menawarkan skema data fleksibel, memberikan kinerja baca/tulis yang sangat baik, dan mendukung kondisi query kompleks. Hologres adalah gudang data all-in-one yang mendukung penulisan dan pembaruan data secara real-time, di mana data dapat langsung di-query setelah ditulis. Ketiga layanan ini bekerja sama dengan mulus untuk memproses berbagai jenis data secara terpusat, memungkinkan pembaruan, query, dan analisis dataset besar secara real-time. Arsitekturnya adalah sebagai berikut:

1. Sebuah pekerjaan Flink membaca stream CDC (Change Data Capture) dari MongoDB. Saat data diperbarui, pekerjaan Flink mengalirkan primary keys (PKs) dari data yang diperbarui ke Kafka menggunakan Upsert Kafka connector.

2. Jika tabel dimensi diperbarui, pekerjaan Flink melakukan lookup join untuk mengidentifikasi catatan yang sesuai di tabel fakta dan menulis PKs dari data yang terpengaruh di tabel fakta ke Kafka.

3. Pekerjaan Flink kedua mengonsumsi PKs dari Kafka, melakukan lookup joins terhadap tabel fakta dan dimensi MongoDB untuk merekonstruksi catatan tabel lebar lengkap, lalu memperbarui data di Hologres.

Advantages

Solusi ini memberikan manfaat berikut:

• Throughput Penulisan Tinggi dan Penyimpanan yang Dapat Diskalakan: ApsaraDB untuk MongoDB unggul dalam beban kerja baca/tulis konkurensi tinggi, terutama dalam kluster sharded. Ini dapat menskalakan performa dan penyimpanannya untuk menampung volume besar penulisan data yang sering, menghilangkan kemungkinan hambatan penulisan dan batasan penyimpanan.

• Propagasi Perubahan yang Efisien: Saat tabel fakta atau dimensi di MongoDB diperbarui, Flink mengidentifikasi dan hanya mengalirkan PKs dari catatan tabel fakta yang terpengaruh ke Kafka. Pendekatan berbasis perubahan ini memastikan pembaruan yang efisien, karena hanya data relevan yang diproses ulang, terlepas dari total volume data.

• Query dan Analisis Real-time: Hologres mendukung pembaruan efisien dengan latensi rendah dan memungkinkan data di-query segera setelah ditulis. Ini memungkinkan analisis data real-time pada tabel yang baru saja diperbarui.

Praktik Langsung

Bagian ini menunjukkan cara menggunakan Flink untuk memproses data pembelian gamer secara real-time, membangun tabel lebar dengan menggabungkan tabel fakta dan dimensi, dan menulis hasilnya ke Hologres untuk query langsung.

Deskripsi solusi

Solusi ini mereplikasi perubahan pada tabel dimensi ke Hologres secara real-time. Proses ini terdiri dari empat langkah berikut:

1. Tangkap perubahan: Tangkap perubahan data real-time dari tabel dimensi MongoDB.

2. Propagasi perubahan tabel dimensi: Saat tabel dimensi berubah, pekerjaan Flink menggunakan lookup join (misalnya, pada game_id) untuk mengidentifikasi catatan yang terpengaruh di tabel fakta MongoDB dan mengekstraksi primary keys mereka, seperti sale_id.

3. Pemicu rekalkulasi: Kirim primary keys ke Kafka untuk memberi tahu pekerjaan hilir tentang pembaruan data yang diperlukan.

4. Pembaruan inkremental: Ambil data terbaru, rekonstruksi tabel lebar, dan upsert data ke Hologres.

Prasyarat

• Sebuah workspace Realtime Compute for Apache Flink telah dibuat. Realtime Compute for Apache Flink harus menggunakan VVR 8.0.5 atau lebih baru. Untuk informasi lebih lanjut, lihat Aktifkan Realtime Compute for Apache Flink.

• Sebuah instance ApsaraDB for MongoDB versi 4.0 atau lebih baru telah dibuat. Untuk informasi lebih lanjut, lihat Buat Instance Kluster Sharded.

• Sebuah instance Hologres eksklusif versi 1.3 atau lebih baru telah dibuat. Untuk informasi lebih lanjut, lihat Beli Instance Hologres.

• Instans ApsaraMQ untuk Kafka telah dibuat. Untuk informasi lebih lanjut, lihat Deploy instans ApsaraMQ untuk Kafka.

• Workspace Realtime Compute for Apache Flink, instance ApsaraDB for MongoDB, instance Hologres, dan instance ApsaraMQ for Kafka berada di VPC yang sama. Jika tidak berada di VPC yang sama, Anda harus membuat koneksi antar-VPC atau mengizinkan Realtime Compute for Apache Flink mengakses layanan lain melalui Internet. Untuk informasi lebih lanjut, lihat Bagaimana Realtime Compute for Apache Flink Mengakses Layanan Lintas VPC? dan Bagaimana Realtime Compute for Apache Flink Mengakses Internet?

• Pengguna RAM atau Peran RAM memiliki akses ke sumber daya yang relevan.

Langkah 1: Persiapkan data

1. Buat database dan tiga tabel di instance ApsaraDB for MongoDB.

   1. Masuk ke instans ApsaraDB untuk MongoDB Anda.

   2. Tambahkan blok CIDR ruang kerja Flink ke daftar putih instans ApsaraDB untuk MongoDB. Untuk detailnya, lihat Konfigurasikan daftar putih untuk instans dan Bagaimana cara mengonfigurasi daftar putih?

   3. Di editor SQL Konsol Data Management (DMS), buat database bernama mongo_test dengan menjalankan perintah berikut:

      use mongo_test;

   4. Buat tabel game_sales, game_dimension, dan platform_dimension di database, dan sisipkan data ke dalam tabel tersebut.

      // Tabel penjualan game (status adalah bidang logis, 1 berarti ada, 0 berarti dihapus)
      db.game_sales.insert(
        [
      	{sale_id:0,game_id:101,platform_id:1,"sale_date":"2024-01-01",units_sold:500,sale_amt:2500,status:1},
        ]
      );
      
      // Tabel dimensi game
      db.game_dimension.insert(
        [
      	{game_id:101,"game_name":"SpaceInvaders","release_date":"2023-06-15","developer":"DevCorp","publisher":"PubInc"},
      	{game_id:102,"game_name":"PuzzleQuest","release_date":"2023-07-20","developer":"PuzzleDev","publisher":"QuestPub"},
      	{game_id:103,"game_name":"RacingFever","release_date":"2023-08-10","developer":"SpeedCo","publisher":"RaceLtd"},
      	{game_id:104,"game_name":"AdventureLand","release_date":"2023-09-05","developer":"Adventure","publisher":"LandCo"},
        ]
      );
      
      // Tabel dimensi platform
      db.platform_dimension.insert(
        [
      	{platform_id:1,"platform_name":"PCGaming","type":"PC"},
      	{platform_id:2,"platform_name":"PlayStation","type":"Console"},
      	{platform_id:3,"platform_name":"Mobile","type":"Mobile"}
        ]
      );

   5. Query data di tabel:

      db.game_sales.find();
      db.game_dimension.find();
      db.platform_dimension.find();

      

2. Buat tabel di Hologres.

   1. Masuk ke Konsol Hologres, klik Instances di bilah navigasi kiri, lalu klik instans Hologres yang ingin Anda akses. Di sudut kanan atas, klik Connect to Instance.

   2. Di bilah navigasi atas, klik Metadata Management lalu klik Create Database. Di jendela pop-up di bidang Database Name, masukkan test, dan atur Policy ke SPM. Untuk informasi lebih lanjut, lihat Buat database.

      

   3. Di bilah navigasi atas, klik SQL Editor. Di sudut kanan atas panel Query samping, klik ikon SQL untuk membuat query SQL. Pilih instans target dan database, dan salin potongan kode berikut untuk membuat tabel bernama game_sales_details.

      CREATE TABLE game_sales_details(
        sale_id INT not null primary key,
        game_id INT,
        platform_id INT,
        sale_date VARCHAR(50),
        units_sold INT,
        sale_amt INT,
        status INT,
        game_name VARCHAR(50),
        release_date VARCHAR(50),
        developer VARCHAR(50),
        publisher VARCHAR(50),
        platform_name VARCHAR(50),
        type VARCHAR(50)
      );

3. Buat topik ApsaraMQ for Kafka.

   1. Masuk ke Konsol ApsaraMQ untuk Kafka. Klik Instances di bilah navigasi kiri, dan klik instans Anda.

   2. Di bilah navigasi kiri, klik Whitelist Management. Tambahkan blok CIDR ruang kerja Flink ke daftar putih baru atau yang sudah ada.

   3. Di bilah navigasi kiri, klik Topics. Pilih Create Topic. Di panel kanan yang muncul, masukkan game_sales_fact di kolom Nama, masukkan deskripsi di kolom Deskripsi, dan gunakan nilai default untuk bidang lainnya. Klik OK.

Langkah 2: Buat pekerjaan aliran

-- Buat tabel Kafka untuk menyimpan dan mengonsumsi primary keys
CREATE TEMPORARY TABLE game_sales_fact
(
  sale_id  INT
  ,PRIMARY KEY (sale_id) NOT ENFORCED
)
WITH (
  'connector' = 'upsert-kafka'
  ,'properties.bootstrap.servers' = '${secret_values.Kafka-hosts}'
  ,'topic' = 'game_sales_fact'
  ,'key.format' = 'json'
  ,'value.format' = 'json'
  ,'properties.group.id' = 'game_sales_fact'
  ,'properties.auto.offset.reset' = 'earliest'
);

-- Buat tabel game_sales
CREATE TEMPORARY TABLE game_sales
(
  `_id`       STRING,
  sale_id     INT,
  game_id     INT,
  platform_id INT,
  sale_date   STRING,
  units_sold  INT,
  sale_amt    INT,
  status      INT,
  PRIMARY KEY (_id) NOT ENFORCED
)
WITH (
  'connector' = 'mongodb',
  'uri' = '${secret_values.MongoDB-URI}',
  'database' = 'mongo_test',
  'collection' = 'game_sales'
);


-- Buat tabel game_dimension
CREATE TEMPORARY TABLE game_dimension
(
  `_id`        STRING,
  game_id      INT,
  game_name    STRING,
  release_date STRING,
  developer    STRING,
  publisher    STRING,
  PRIMARY KEY (_id) NOT ENFORCED
)
WITH (
  'connector' = 'mongodb',
  'uri' = '${secret_values.MongoDB-URI}',
  'database' = 'mongo_test',
  'collection' = 'game_dimension'
);

-- Buat tabel platform_dimension
CREATE TEMPORARY TABLE platform_dimension
(
  `_id`          STRING
  ,platform_id   INT
  ,platform_name STRING
  ,type          STRING
  ,PRIMARY KEY (_id) NOT ENFORCED
)
WITH (
  'connector' = 'mongodb',
  'uri' = '${secret_values.MongoDB-URI}',
  'database' = 'mongo_test',
  'collection' = 'platform_dimension'
);

-- Buat tabel sink game_sales_details
CREATE TEMPORARY TABLE IF NOT EXISTS game_sales_details
(
  sale_id       INT,
  game_id       INT,
  platform_id   INT,
  sale_date     STRING,
  units_sold    INT,
  sale_amt      INT,
  status        INT,
  game_name     STRING,
  release_date  STRING,
  developer     STRING,
  publisher     STRING,
  platform_name STRING,
  type          STRING,
  PRIMARY KEY (sale_id) NOT ENFORCED
)
WITH (
  -- Opsi konektor Hologres untuk VVR 11.
  'connector' = 'hologres',
  'dbname' = 'test', -- Nama database Hologres
  'tablename' = 'public.game_sales_details', -- Nama tabel Hologres
  'username' = '${secret_values.AccessKeyID}', -- ID AccessKey akun Alibaba Cloud Anda
  'password' = '${secret_values.AccessKeySecret}', -- Rahasia AccessKey akun Alibaba Cloud Anda
  'endpoint' = '${secret_values.Hologres-endpoint}', -- Titik akhir VPC instans Hologres Anda
  'sink.delete-strategy'='IGNORE_DELETE',       -- Kebijakan untuk menangani pesan yang ditarik kembali. IGNORE_DELETE hanya menyisipkan atau memperbarui data, tidak menghapusnya.
  'sink.on-conflict-action'='INSERT_OR_UPDATE', -- Mode pembaruan. Aktifkan ini untuk melakukan pembaruan kolom parsial.
  'sink.partial-insert.enabled'='true'
);

INSERT INTO game_sales_details (
  sale_id,
  game_id,
  platform_id,
  sale_date,
  units_sold,
  sale_amt,
  status,
  game_name,
  release_date,
  developer,
  publisher,
  platform_name,
  type
)
select
  gsf.sale_id,
  gs.game_id,
  gs.platform_id,
  gs.sale_date,
  gs.units_sold,
  gs.sale_amt,
  gs.status,
  gd.game_name,
  gd.release_date,
  gd.developer,
  gd.publisher,
  pd.platform_name,
  pd.type
from game_sales_fact as gsf
join game_sales FOR SYSTEM_TIME AS OF PROCTIME() as gs
on gsf.sale_id = gs.sale_id

join game_dimension FOR SYSTEM_TIME AS OF PROCTIME() as gd
on gs.game_id = gd.game_id

join platform_dimension FOR SYSTEM_TIME AS OF PROCTIME() as pd
on gs.platform_id = pd.platform_id;

Langkah 3: Mulai pekerjaan

1. Di menu navigasi kiri Konsol Pengembangan, pilih O&M > Deployments. Di halaman Deployments, mulai dua penyebaran pekerjaan.

2. Setelah status pekerjaan berubah menjadi running, buka HoloWeb untuk query tabel game_sales_details.

   SELECT * FROM game_sales_details;

   Anda dapat melihat bahwa baris data baru dimasukkan ke dalam tabel.

   

Langkah 4: Perbarui dan query data

Pembaruan pada tabel game_sales dan tabel dimensi di MongoDB disinkronkan ke Hologres untuk query dan analisis. Contoh berikut menunjukkan bagaimana data diperbarui secara real-time.

Tabel game_sales diperbarui

1. Masukkan lima baris ke tabel game_sales di MongoDB:

   db.game_sales.insert(
     [
   	{sale_id:1,game_id:101,platform_id:1,"sale_date":"2024-01-01",units_sold:500,sale_amt:2500,status:1},
   	{sale_id:2,game_id:102,platform_id:2,"sale_date":"2024-08-02",units_sold:400,sale_amt:2000,status:1},
   	{sale_id:3,game_id:103,platform_id:1,"sale_date":"2024-08-03",units_sold:300,sale_amt:1500,status:1},
   	{sale_id:4,game_id:101,platform_id:3,"sale_date":"2024-08-04",units_sold:200,sale_amt:1000,status:1},
   	{sale_id:5,game_id:104,platform_id:2,"sale_date":"2024-08-05",units_sold:100,sale_amt:3000,status:1}
     ]
   );

   Query tabel game_sales_details di Hologres, dan Anda dapat melihat baris yang sesuai telah dimasukkan.

   

2. Ubah nilai bidang sale_date di tabel game_sales MongoDB dari '2024-01-01' menjadi '2024-08-01'.

   db.game_sales.updateMany({"sale_date": "2024-01-01"}, {$set: {"sale_date": "2024-08-01"}});

   Query tabel game_sales_details di Hologres. Anda dapat menemukan bahwa nilai bidang sale_date telah berubah dari '2024-01-01' menjadi '2024-08-01'.

   

3. Lakukan penghapusan logis pada tabel game_sales di ApsaraDB untuk MongoDB dengan mengubah nilai status menjadi 0 untuk baris di mana sale_id adalah 5.

   db.game_sales.updateMany({"sale_id": 5}, {$set: {"status": 0}});

   Query tabel game_sales_details di Hologres. Anda dapat melihat bahwa nilai kolom status untuk catatan di mana sale_id adalah 5 telah berubah menjadi 0. Ini menyelesaikan penghapusan logis.

   

Pembaruan tabel dimensi

1. Masukkan baris baru ke tabel game_dimension dan platform_dimension.

   // Tabel dimensi game
   db.game_dimension.insert(
     [
   	{game_id:105,"game_name":"HSHWK","release_date":"2024-08-20","developer":"GameSC","publisher":"GameSC"},
   	{game_id:106,"game_name":"HPBUBG","release_date":"2018-01-01","developer":"BLUE","publisher":"KK"}
     ]
   );
   
   // Tabel dimensi platform
   db.platform_dimension.insert(
     [
   	{platform_id:4,"platform_name":"Steam","type":"PC"},
   	{platform_id:5,"platform_name":"Epic","type":"PC"}
     ]
   );

   Memasukkan data baru ke tabel dimensi saja tidak memicu sinkronisasi data dari MongoDB ke Hologres. Proses sinkronisasi ini dipicu oleh perubahan pada tabel game_sales, yang dihasilkan dari pembelian atau unduhan. Oleh karena itu, Anda perlu memasukkan data ke tabel game_sales.

   // Tabel penjualan game
   db.game_sales.insert(
     [
   	{sale_id:6,game_id:105,platform_id:4,"sale_date":"2024-09-01",units_sold:400,sale_amt:2000,,status:1},
   	{sale_id:7,game_id:106,platform_id:1,"sale_date":"2024-09-01",units_sold:300,sale_amt:1500,,status:1}
     ]
   );

   Verifikasi pembaruan dengan men-query tabel game_sales_details di Hologres. Anda dapat menemukan bahwa dua catatan data baru telah dimasukkan.

   

2. Perbarui data di tabel game_dimension dan platform_dimension MongoDB.

   // Perbarui tanggal rilis
   db.game_dimension.updateMany({"release_date": "2018-01-01"}, {$set: {"release_date": "2024-01-01"}});
   
   // Perbarui tipe platform
   db.platform_dimension.updateMany({"type": "PC"}, {$set: {"type": "Swich"}});

   Anda dapat menemukan bahwa bidang terkait diperbarui di tabel Hologres.

   

Referensi

• Bangun Gudang Data Real-Time Menggunakan Realtime Compute for Apache Flink dan Hologres

• Bangun danau data streaming menggunakan Paimon dan StarRocks

• Versi MongoDB Mana yang Didukung oleh Konektor MongoDB CDC?

• Dapatkah Konektor MongoDB CDC Membaca Data dari Checkpoint Deployment Setelah Deployment Di-Restart? Apa Prinsip Kerjanya?