Sort and aggregate data with a UDAF - Realtime Compute for Apache Flink

Örnek veri

electric_info tablosu konut elektrik şebekesi terminallerinden gelen verileri içerir. Tabloda olay kimliği (event_id), kullanıcı kimliği (user_id), olay zamanı (event_time) ve terminal durumu (status) bulunur. Her kullanıcı için status değerlerini toplayacak ve bunları event_time alanına göre sıralayacaksınız.

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

Beklenen sonuç

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

Adım 1: Veri kaynağını hazırlama

Bu örnekte veri kaynağı olarak ApsaraDB RDS kullanılmaktadır.

ApsaraDB RDS for MySQL örneği oluşturma.

Not
ApsaraDB RDS for MySQL örneğiniz, Realtime Compute for Apache Flink çalışma alanınızla aynı VPC içinde olmalıdır. Farklı VPC'lerde bulunuyorlarsa bkz. Ağ bağlantısı.
Veritabanı ve hesap oluşturma.

electric adında bir veritabanı ve bu veritabanı için okuma ve yazma izinlerine sahip bir hesap oluşturun.

Data Management (DMS) kullanarak ApsaraDB RDS for MySQL örneğinde oturum açma işlemine gidin, electric veritabanında electric_info ve electric_info_SortListAgg tablolarını oluşturun ve veri ekleyin.

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');

Adım 2: UDAF'i kaydetme

ASI_UDX-1,0-SNAPSHOT.jar paketini indirin.

pom.xml dosyası, Flink 1.17.1 sürümündeki bu özel işlev için gereken minimum bağımlılıklarla yapılandırılmıştır. Özel işlevler hakkında daha fazla bilgi için bkz. Özel işlevler.

Örnek ASI_UDAF kodu birden çok satırı tek bir satırda birleştirir ve verileri belirtilen bir sütuna göre sıralar. Kodu iş ihtiyaçlarınıza uyacak şekilde değiştirebilirsiniz.

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) {
		}
	}
}

UDAF'i kaydedin.

Bir UDAF'i kaydetmek, kodunu diğer işlerde yeniden kullanmanıza olanak tanır. Java UDAF'leri için JAR dosyasını bir bağımlılık dosyası olarak da yükleyebilirsiniz. Daha fazla bilgi için bkz. Kullanıcı tanımlı toplama işlevleri (UDAF).
1. Realtime Compute for Apache Flink konsolu üzerinde oturum açın.
2. Hedef çalışma alanını bulun ve Actions sütunundaki Console seçeneğine tıklayın.
3. Sol gezinme bölmesinde Development > ETL yolunu izleyin.
4. UDFs sekmesinde Register UDF Artifact seçeneğine tıklayın.
Click to select bölümünde 1. Adımdaki JAR dosyasını yükleyin ve OK seçeneğine tıklayın.

İletişim kutusu Upload File ve External URL olmak üzere iki kayıt yöntemi sunar. Ayrıca bir UDF Name belirtmeniz gerekir ve isteğe bağlı olarak bir dependency file yükleyebilirsiniz.
Not
- UDF'inizin JAR dosyası çalışma alanıyla ilişkili OSS Bucket'ın sql-artifacts dizinine yüklenir.
- Realtime Compute for Apache Flink konsolu UDF JAR dosyanızı ayrıştırır ve Flink UDF, UDAF ve UDTF arayüzlerini kullanan sınıfları algılar. Sınıf adlarını otomatik olarak çıkarır ve Function Name alanını doldurur.
Manage Functions iletişim kutusunda Create Functions seçeneğine tıklayın.

Kaydedilen UDF, SQL düzenleyici sayfasının sol tarafındaki UDFs listesinde görünür.

Adım 3: Flink işi oluşturma

Development > ETL sayfasında New seçeneğine tıklayın.
Blank Stream Draft seçeneğine tıklayın.
Next seçeneğine tıklayın.

New Draft iletişim kutusunda iş ayarlarını yapılandırın.

Parameter	Description
Name	İş için benzersiz bir ad. Not İş adı, mevcut proje içinde benzersiz olmalıdır.
Location	İşin depolama konumu. Mevcut bir klasörün yanındaki simgesine tıklayarak alt klasör de oluşturabilirsiniz.
Engine version	İş için Flink altyapı sürümü. Bu sürüm, `pom.xml` dosyanızda belirtilen sürümle eşleşmelidir. Altyapı sürümleri, sürüm eşlemeleri ve yaşam döngüsü bilgileri hakkında ayrıntılı bilgi için bkz. Engine Versions.

DDL ve DML deyimlerini yazın.

-- 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;

Aşağıdaki tabloda parametreler açıklanmıştır. Bunları gerçek ihtiyaçlarınıza göre değiştirin. MySQL konektör parametreleri hakkında daha fazla bilgi için bkz. MySQL konektörü.

Parameter	Description	Notes
connector	Konektör türü.	Bu örnekte değer `mysql` olarak sabittir.
hostname	MySQL veritabanının IP adresi veya ana bilgisayar adı.	Bu örnekte ApsaraDB RDS for MySQL örneğinin iç uç noktası kullanılır.
username	MySQL veritabanı hizmeti için kullanıcı adı.	Yok.
password	MySQL veritabanı hizmeti için parola.	Bu örnekte güvenlik risklerini önlemek amacıyla parola için `mysql_pw` adlı bir değişken kullanılır. Daha fazla bilgi için bkz. Değişkenler.
database-name	MySQL veritabanının adı.	Bu örnekte Adım 1: Veri kaynağını hazırlama bölümünde oluşturulan `electric` veritabanı kullanılır.
table-name	MySQL tablosunun adı.	Bu örnekte bunu `electric_info` veya `electric_info_sortlistagg` olarak ayarlayın.
port	MySQL veritabanı hizmetinin bağlantı noktası numarası.	Yok.

(İsteğe bağlı) Sağ üst köşede Validate ve Debug seçeneklerine tıklayın. Bu özellikler hakkında daha fazla bilgi için bkz. İş geliştirme genel bakışı.
Deploy seçeneğine ve ardından Confirm seçeneğine tıklayın.
O&M > Deployments sayfasında hedef işi bulun, Actions sütunundaki Start seçeneğine tıklayın ve Initial Mode seçeneğini belirleyin.

Adım 4: Sonucu sorgulama

ApsaraDB RDS'de aşağıdaki deyimi çalıştırarak toplanmış ve sıralanmış sonuçları görüntüleyin.

SELECT * FROM `electric_info_sortlistagg`;

Çıktı, her kullanıcı için durumların doğru şekilde toplandığını ve sıralandığını doğrular: user_id=1222 değeri status_sort=TD,LD,TD,TD değerine karşılık gelir; user_id=1333 değeri status_sort=TD,LD,LD,LD değerine karşılık gelir ve user_id=1555 değeri status_sort=TD,LD değerine karşılık gelir.

İlgili bilgiler

Flink'in desteklediği yerleşik işlevler hakkında bilgi için bkz. Yerleşik işlevler.
İşleri dağıtma ve başlatma hakkında bilgi için bkz. İşi dağıtma ve İşi başlatma.
İş çalıştırma parametrelerini değiştirmek için bkz. İş dağıtımını yapılandırma. Bazı parametreler, işlerin durdurulup yeniden başlatılmasından kaynaklanan kesinti süresini azaltmak için dinamik olarak güncellenebilir. Daha fazla bilgi için bkz. Dinamik ölçeklendirme ve parametre güncellemeleri.
SQL işlerinde Python özel işlevlerini kullanma hakkında bilgi için bkz. Özel işlevler.