Realtime Compute for Apache Flink:Materialized Table (Build a Unified Stream-Batch Lakehouse)

1. Buat Paimon Catalog.

   Tabel materialisasi didukung oleh Apache Paimon. Anda harus membuat Paimon Catalog dengan tipe metastore Filesystem. Lewati langkah ini jika Anda sudah memilikinya. Untuk informasi lebih lanjut, lihat Create a Paimon Catalog.

   Create a Paimon Catalog

   1. Masuk ke Realtime Compute Management Console.

   2. Klik Console pada kolom Actions ruang kerja target Anda.

   3. Pada panel navigasi kiri, pilih Data Management dan klik Create Catalog. Pilih Apache Paimon, lalu klik Next.

      Parameter description:

      Configuration item	Description	Remarks
      metastore	Metastore type.	Contoh ini menggunakan filesystem sebagai tipe metastore.
      catalog name	Nama Paimon Catalog.	Masukkan nama kustom dalam bahasa Inggris. Contoh ini menggunakan paimon.
      warehouse	Direktori gudang data di OSS.	Gunakan format oss://<bucket>/<object>. Di mana: <bucket>: nama Bucket OSS Anda. <object>: path tempat data Anda disimpan. Periksa nama bucket dan object Anda di OSS Management Console.
      fs.oss.endpoint	Alamat koneksi OSS.	Jika Flink dan OSS berada di Wilayah yang sama, gunakan titik akhir jaringan pribadi. Jika tidak, gunakan titik akhir jaringan publik. Untuk informasi lebih lanjut, lihat Regions and endpoints.
      fs.oss.accessKeyId	ID AccessKey Akun Alibaba Cloud atau Pengguna RAM Anda yang memiliki izin baca dan tulis pada OSS.	Untuk petunjuk cara memperolehnya, lihat Create an AccessKey. Untuk menghindari eksposur kredensial dalam teks biasa, gunakan variabel sebagai gantinya. Untuk informasi lebih lanjut, lihat Variable management.
      fs.oss.accessKeySecret	Rahasia AccessKey Akun Alibaba Cloud atau Pengguna RAM Anda yang memiliki izin baca dan tulis pada OSS.

2. Buat tabel log perilaku pengguna ods_user_log dan tabel informasi produk ods_dim_product.

   1. Masuk ke Realtime Compute Management Console.

   2. Klik Console pada kolom Actions ruang kerja target Anda.

   3. Pada panel navigasi kiri, pilih Data Development > Data Query. Salin dan tempel kode berikut untuk membuat tabel sumber.

      Contoh ini mengasumsikan Anda telah membuat Paimon Catalog bernama paimon dan menggunakan database default.

      CREATE TABLE `paimon`.`default`.`ods_user_log` (
        item_id INT NOT NULL,
        user_id INT NOT NULL,
        vtime TIMESTAMP(6),
        ds VARCHAR(10) NOT NULL
      ) 
      PARTITIONED BY(ds)
      WITH (
        'bucket' = '4',            -- Setel jumlah bucket menjadi 4.
        'bucket-key' = 'item_id'   -- Tentukan kunci yang digunakan untuk menentukan penempatan bucket. Baris dengan item_id yang sama masuk ke bucket yang sama.
      );
      CREATE TABLE `paimon`.`default`.`ods_dim_product` (
        item_id INT NOT NULL,
        title VARCHAR(255),
        pict_url VARCHAR(255), 
        brand_id INT,
        seller_id INT,
        PRIMARY KEY(item_id) NOT ENFORCED
      ) WITH (
        'bucket' = '4',
        'bucket-key' = 'item_id'
      );

   4. Klik Run di pojok kanan atas untuk membuat tabel.

   5. Pada panel navigasi kiri, pilih Data Management, klik Paimon Catalog Anda, lalu klik Refresh untuk melihat tabel baru.

3. Gunakan konektor Faker untuk pembuatan data simulasi guna menghasilkan data tiruan dan menuliskannya ke tabel Paimon.

   1. Pada panel navigasi kiri, pilih Data Development > ETL.

   2. Klik New, pilih Blank stream draft, klik Next, lalu klik Create.

   3. Salin pernyataan SQL berikut ke editor SQL.

      CREATE TEMPORARY TABLE `user_log` (
        item_id INT,  // Product ID
        user_id INT,  // User ID
        vtime TIMESTAMP,  
        ds AS DATE_FORMAT(CURRENT_DATE,'yyyyMMdd')
      ) WITH (
        'connector' = 'faker',    -- Faker connector
        'fields.item_id.expression'='#{number.numberBetween ''0'',''1000''}',    -- Hasilkan bilangan acak antara 0 dan 1000
        'fields.user_id.expression'='#{number.numberBetween ''0'',''100''}',
        'fields.vtime.expression'='#{date.past ''5'',''HOURS''}',           -- Hasilkan data hingga 5 jam sebelum waktu saat ini
        'rows-per-second' = '3'   -- Hasilkan 3 baris per detik
       );
       CREATE TEMPORARY TABLE `dim_product` (
        item_id INT NOT NULL,
        title VARCHAR(255),
        pict_url VARCHAR(255), 
        brand_id INT,
        seller_id INT,
        PRIMARY KEY(item_id) NOT ENFORCED
       ) WITH (
        'connector' = 'faker',    -- Faker connector
        'fields.item_id.expression'='#{number.numberBetween ''0'',''1000''}',
        'fields.title.expression'='#{book.title}',
        'fields.pict_url.expression'='#{internet.domainName}',
        'fields.brand_id.expression'='#{number.numberBetween ''1000'',''10000''}',   
        'fields.seller_id.expression'='#{number.numberBetween ''1000'',''10000''}',
        'rows-per-second' = '3'        -- Hasilkan 3 baris per detik
       );
      BEGIN STATEMENT SET; 
      INSERT INTO `paimon`.`default`.`ods_user_log` 
        SELECT 
        item_id,
        user_id,
        vtime,
        CAST(ds AS VARCHAR(10)) AS ds
      FROM `user_log`;
      INSERT INTO `paimon`.`default`.`ods_dim_product`
        SELECT 
        item_id,
        title,
        pict_url,
        brand_id,
        seller_id
      FROM `dim_product`;
      END; 

   4. Klik Deploy di pojok kanan atas untuk men-deploy pekerjaan.

   5. Pada panel navigasi kiri, pilih Operation Center > Job O&M. Klik Start pada kolom Actions pekerjaan target Anda, pilih Stateless start, lalu klik Start.

4. Kueri data simulasi.

   Pada panel navigasi kiri, pilih Data Development > Data Query. Salin pernyataan SQL berikut ke editor SQL dan klik Run di pojok kanan atas.

   SELECT * FROM `paimon`.`default`.ods_dim_product LIMIT 10;
   SELECT * FROM `paimon`.`default`.ods_user_log LIMIT 10;

Pratinjau data

Anda dapat melihat pratinjau 100 baris terbaru dari tabel materialisasi.

1. Pada panel navigasi kiri, pilih Operation Center > Data Lineage dan cari tabel materialisasi target Anda.

2. Klik tampilan tabel materialisasi tersebut, lalu klik Details di pojok kanan bawah halaman.

3. Pada tab Data Preview tabel materialisasi, klik ikon Query.

Kueri data

Pada panel navigasi kiri, pilih Data Development > Data Query. Salin pernyataan SQL berikut ke editor SQL, pilih potongan kode tersebut, lalu klik Run untuk mengkueri tabel materialisasi dws_overall.

SELECT * FROM `paimon`.`default`.dws_overall ORDER BY hh;