Realtime Compute for Apache Flink:Urutkan dan agregasi data dengan UDAF

Data contoh

Tabel electric_info berisi data dari terminal jaringan listrik perumahan, mencakup ID event (event_id), ID pengguna (user_id), waktu event (event_time), dan status terminal (status). Anda akan mengagregasi nilai status untuk setiap pengguna dan mengurutkannya berdasarkan event_time.

• electric_info

  event_id	user_id	event_time	status
  1	1222	2023-06-30 11:14:00	LD
  2	1333	2023-06-30 11:12:00	LD
  3	1222	2023-06-30 11:11:00	TD
  4	1333	2023-06-30 11:12:00	LD
  5	1222	2023-06-30 11:15:00	TD
  6	1333	2023-06-30 11:18:00	LD
  7	1222	2023-06-30 11:19:00	TD
  8	1333	2023-06-30 11:10:00	TD
  9	1555	2023-06-30 11:16:00	TD
  10	1555	2023-06-30 11:17:00	LD

• Hasil yang diharapkan

  user_id	status
  1222	TD,LD,TD,TD
  1333	TD,LD,LD,LD
  1555	TD,LD

Langkah 1: Siapkan sumber data

Contoh ini menggunakan ApsaraDB RDS sebagai sumber data.

1. Buat instans ApsaraDB RDS for MySQL.

   Catatan

   Instans ApsaraDB RDS for MySQL Anda harus berada dalam VPC yang sama dengan ruang kerja Realtime Compute for Apache Flink Anda. Jika berada di VPC berbeda, lihat Konektivitas jaringan.

2. Buat database dan akun.

   Buat database bernama electric dan akun dengan izin baca-tulis untuk database tersebut.

3. Masuk ke instans ApsaraDB RDS for MySQL menggunakan Data Management (DMS), buat tabel electric_info dan electric_info_SortListAgg di database electric, lalu masukkan data.

   CREATE TABLE `electric_info` (
     event_id bigint NOT NULL PRIMARY KEY COMMENT 'Event ID',
     user_id bigint NOT NULL COMMENT 'User ID', 
     event_time timestamp NOT NULL COMMENT 'Event time',
     status varchar(10) NOT NULL COMMENT 'User terminal status'
   );
   CREATE TABLE `electric_info_SortListAgg` (
     user_id bigint NOT NULL PRIMARY KEY COMMENT 'User ID', 
     status_sort varchar(50) NULL COMMENT 'User terminal status sorted in ascending order by event time'
   );
   -- Prepare data
   INSERT INTO electric_info VALUES 
   (1,1222,'2023-06-30 11:14','LD'),
   (2,1333,'2023-06-30 11:12','LD'),
   (3,1222,'2023-06-30 11:11','TD'),
   (4,1333,'2023-06-30 11:12','LD'),
   (5,1222,'2023-06-30 11:15','TD'),
   (6,1333,'2023-06-30 11:18','LD'),
   (7,1222,'2023-06-30 11:19','TD'),
   (8,1333,'2023-06-30 11:10','TD'),
   (9,1555,'2023-06-30 11:16','TD'),
   (10,1555,'2023-06-30 11:17','LD');

Langkah 2: Daftarkan UDAF

1. Unduh paket ASI_UDX-1.0-SNAPSHOT.jar.

   File pom.xml dikonfigurasi dengan dependensi minimum yang diperlukan untuk fungsi kustom ini pada Flink versi 1.17.1. Untuk informasi selengkapnya tentang fungsi kustom, lihat Fungsi kustom.

2. Kode contoh ASI_UDAF menggabungkan beberapa baris menjadi satu baris dan mengurutkan data berdasarkan kolom tertentu. Anda dapat memodifikasi kode ini sesuai kebutuhan bisnis Anda.

   package ASI_UDAF;
   import org.apache.commons.lang3.StringUtils;
   import org.apache.flink.table.functions.AggregateFunction;
   import java.util.ArrayList;
   import java.util.Comparator;
   import java.util.Iterator;
   import java.util.List;
   public class ASI_UDAF{
   	/**Accumulator class*/
   	public static class AcList {
   		public  List<String> list;
   	}
   	/**Aggregate function class*/
   	public static class SortListAgg extends AggregateFunction<String,AcList> {
   		public String getValue(AcList asc) {
   			/**Sort the data in the list based on a specific rule*/
   			asc.list.sort(new Comparator<String>() {
   				@Override
   				public int compare(String o1, String o2) {
   					return Integer.parseInt(o1.split("#")[1]) - Integer.parseInt(o2.split("#")[1]);
   				}
   			});
   			/**Traverse the sorted list, extract the required fields, and join them into a string*/
   			List<String> ret = new ArrayList<String>();
   			Iterator<String> strlist = asc.list.iterator();
   			while (strlist.hasNext()) {
   				ret.add(strlist.next().split("#")[0]);
   			}
   			String str = StringUtils.join(ret, ',');
   			return str;
   		}
   		/**Method to create an accumulator*/
   		public AcList createAccumulator() {
   			AcList ac = new AcList();
   			List<String> list = new ArrayList<String>();
   			ac.list = list;
   			return ac;
   		}
   		/**Accumulation method: Add the input data to the accumulator*/
   		public void accumulate(AcList acc, String tuple1) {
   			acc.list.add(tuple1);
   		}
   		/**Retraction method*/
   		public void retract(AcList acc, String num) {
   		}
   	}
   }

3. Daftarkan UDAF.

   Pendaftaran UDAF memungkinkan Anda menggunakan kembali kodenya di pekerjaan lain. Untuk UDAF Java, Anda juga dapat mengunggah file JAR sebagai file dependensi. Untuk informasi selengkapnya, lihat Fungsi agregat yang didefinisikan pengguna (UDAF).

   1. Masuk ke Konsol Realtime Compute for Apache Flink.

   2. Temukan ruang kerja target dan klik Console di kolom Actions.

   3. Di panel navigasi kiri, pilih Development > ETL.

   4. Pada tab Functions, klik Register UDF.

4. Pada bagian Select File, unggah file JAR dari Langkah 1 dan klik OK.

   Kotak dialog menyediakan dua metode pendaftaran: Upload File dan External URL. Anda juga harus menentukan UDF Name dan secara opsional dapat mengunggah dependency file.

   Catatan

   • File JAR UDF Anda diunggah ke direktori sql-artifacts bucket OSS yang dikaitkan dengan ruang kerja.

   • Konsol Realtime Compute for Apache Flink mengurai file JAR UDF Anda dan mendeteksi kelas yang menggunakan antarmuka Flink UDF, UDAF, dan UDTF. Sistem secara otomatis mengekstraksi nama kelas dan mengisi bidang Function Name.

5. Pada kotak dialog Manage Functions, klik Create Function.

   UDF yang terdaftar muncul di daftar Functions di sisi kiri halaman editor SQL.

Langkah 3: Buat pekerjaan Flink

1. Pada halaman Development > ETL, klik New.

2. Klik Blank Stream Draft.

3. Klik Next.

4. Pada kotak dialog New Draft, konfigurasikan pengaturan pekerjaan.

   Parameter	Description
   File name	Nama unik untuk pekerjaan. Catatan Nama pekerjaan harus unik dalam proyek saat ini.
   Storage location	Lokasi penyimpanan untuk pekerjaan. Anda juga dapat mengklik ikon di sebelah folder yang ada untuk membuat subfolder.
   Engine version	Versi engine Flink untuk pekerjaan. Ini harus sesuai dengan versi yang ditentukan dalam file pom.xml Anda. Untuk detail tentang versi engine, pemetaan versi, dan informasi siklus hidup, lihat Engine Versions.

5. Tulis pernyataan DDL dan DML.

   -- Create the temporary table electric_info.
   CREATE TEMPORARY TABLE electric_info (
     event_id bigint not null,
     `user_id` bigint not null, 
     event_time timestamp(6) not null,
     status string not null,
     primary key(event_id) not enforced
   ) WITH (
     'connector' = 'mysql',
     'hostname' = 'rm-bp1s1xgll21******.mysql.rds.aliyuncs.com',
     'port' = '3306',
     'username' = 'your_username',
     'password' = '${secret_values.mysql_pw}',
     'database-name' = 'electric',
     'table-name' = 'electric_info'
   );
   CREATE TEMPORARY TABLE electric_info_sortlistagg (
     `user_id` bigint not null, 
     status_sort varchar(50) not null,
     primary key(user_id) not enforced
   ) WITH (
     'connector' = 'mysql',
     'hostname' = 'rm-bp1s1xgll21******.mysql.rds.aliyuncs.com',
     'port' = '3306',
     'username' = 'your_username',
     'password' = '${secret_values.mysql_pw}',
     'database-name' = 'electric',
     'table-name' = 'electric_info_sortlistagg'
   );
   -- Aggregate data from the electric_info table and insert it into the electric_info_sortlistagg table.
   -- Pass a concatenated string of status and event_time as a parameter to the registered custom function ASI_UDAF$SortListAgg.
   INSERT INTO electric_info_sortlistagg 
   SELECT `user_id`, `ASI_UDAF$SortListAgg`(CONCAT(status,'#',CAST(UNIX_TIMESTAMP(event_time) as STRING)))
   FROM electric_info GROUP BY user_id;

   Tabel berikut menjelaskan parameter tersebut. Sesuaikan berdasarkan kebutuhan aktual Anda. Untuk informasi selengkapnya tentang parameter konektor MySQL, lihat MySQL connector.

   Parameter	Description	Notes
   connector	Jenis konektor.	Dalam contoh ini, nilainya tetap mysql.
   hostname	Alamat IP atau hostname database MySQL.	Contoh ini menggunakan titik akhir internal instans ApsaraDB RDS for MySQL.
   username	Username untuk layanan database MySQL.	Tidak ada.
   password	Password untuk layanan database MySQL.	Contoh ini menggunakan variabel bernama mysql_pw untuk password guna menghindari risiko keamanan. Untuk informasi selengkapnya, lihat Variables.
   database-name	Nama database MySQL.	Contoh ini menggunakan electric, yaitu database yang dibuat di Langkah 1: Siapkan sumber data.
   table-name	Nama tabel MySQL.	Dalam contoh ini, atur ke electric_info atau electric_info_sortlistagg.
   port	Nomor port layanan database MySQL.	Tidak ada.

6. (Opsional) Di pojok kanan atas, klik Validate dan Debug. Untuk informasi selengkapnya tentang fitur-fitur ini, lihat Job development overview.

7. Klik Deploy, lalu klik Confirm.

8. Pada halaman O&M > Deployments, temukan pekerjaan target, klik Start di kolom Actions, lalu pilih Initial Mode.

Langkah 4: Kueri hasil

Di ApsaraDB RDS, jalankan pernyataan berikut untuk melihat hasil agregasi dan pengurutan.

SELECT * FROM `electric_info_sortlistagg`;

Output mengonfirmasi bahwa status untuk setiap pengguna berhasil diagregasi dan diurutkan: user_id=1222 bersesuaian dengan status_sort=TD,LD,TD,TD; user_id=1333 bersesuaian dengan status_sort=TD,LD,LD,LD; dan user_id=1555 bersesuaian dengan status_sort=TD,LD.

Referensi

• Untuk informasi tentang fungsi bawaan yang didukung Flink, lihat Built-in functions.

• Untuk informasi tentang penerapan dan menjalankan pekerjaan, lihat Deploy a job dan Start a job.

• Untuk memodifikasi parameter runtime pekerjaan, lihat Configure job deployment. Beberapa parameter dapat diperbarui secara dinamis untuk mengurangi downtime akibat menghentikan dan menjalankan ulang pekerjaan. Untuk informasi selengkapnya, lihat Dynamic scaling and parameter updates.

• Untuk informasi tentang penggunaan fungsi kustom Python dalam pekerjaan SQL, lihat Custom functions.