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

Sample data

The electric_info table contains data from residential power grid terminals. It includes the event ID (event_id), user ID (user_id), event time (event_time), and terminal status (status). You will aggregate the status values for each user and sort them by 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

Expected result

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

Step 1: Prepare the data source

This example uses ApsaraDB RDS as the data source.

Create an ApsaraDB RDS for MySQL instance.

Note
Your ApsaraDB RDS for MySQL instance must be in the same VPC as your Realtime Compute for Apache Flink workspace. If they are in different VPCs, see Network connectivity.
Create a database and an account.

Create a database named electric and an account with read and write permissions for this database.

Log on to the ApsaraDB RDS for MySQL instance using Data Management (DMS), create the electric_info and electric_info_SortListAgg tables in the electric database, and insert 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');

Step 2: Register the UDAF

Download the ASI_UDX-1.0-SNAPSHOT.jar package.

The pom.xml file is configured with the minimum dependencies required for this custom function in Flink version 1.17.1. For more information about custom functions, see Custom functions.

The sample ASI_UDAF code merges multiple rows into a single row and sorts the data by a specified column. You can modify the code to fit your business needs.

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

Register the UDAF.

Registering a UDAF lets you reuse its code in other jobs. For Java UDAFs, you can also upload the JAR as a dependency file. For more information, see User-defined aggregate functions (UDAFs).
1. Log on to the Realtime Compute for Apache Flink console.
2. Find the target workspace and click Console in the Actions column.
3. In the left-side navigation pane, choose Development > ETL.
4. On the Functions tab, click Register UDF.
In the Select File section, upload the JAR file from Step 1 and click OK.

The dialog box provides two registration methods: Upload File and External URL. You must also specify a UDF Name and can optionally upload a dependency file.
Note
- The JAR file of your UDF is uploaded to the sql-artifacts directory of the OSS Bucket that is associated with the workspace.
- The Realtime Compute for Apache Flink console parses your UDF JAR file and detects classes that use Flink UDF, UDAF, and UDTF interfaces. It automatically extracts the class names and populates the Function Name field.
In the Manage Functions dialog box, click Create Function.

The registered UDF appears in the Functions list on the left side of the SQL editor page.

Step 3: Create a Flink job

On the Development > ETL page, click New.
Click Blank Stream Draft.
Click Next.

In the New Draft dialog box, configure the job settings.

Parameter	Description
File name	A unique name for the job. Note The job name must be unique within the current project.
Storage location	The storage location for the job. You can also click the icon next to an existing folder to create a subfolder.
Engine version	The Flink engine version for the job. This must match the version specified in your `pom.xml` file. For details about engine versions, version mappings, and lifecycle information, see Engine Versions.

Write the DDL and DML statements.

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

The following table describes the parameters. Modify them based on your actual needs. For more information about MySQL connector parameters, see MySQL connector.

Parameter	Description	Notes
connector	The type of connector.	In this example, the value is fixed to `mysql`.
hostname	The IP address or hostname of the MySQL database.	This example uses the internal endpoint of the ApsaraDB RDS for MySQL instance.
username	The username for the MySQL database service.	None.
password	The password for the MySQL database service.	This example uses a variable named `mysql_pw` for the password to avoid security risks. For more information, see Variables.
database-name	The name of the MySQL database.	This example uses `electric`, the database created in Step 1: Prepare the data source.
table-name	The name of the MySQL table.	In this example, set this to `electric_info` or `electric_info_sortlistagg`.
port	The port number of the MySQL database service.	None.

(Optional) In the upper-right corner, click Validate and Debug. For more information about these features, see Job development overview.
Click Deploy, and then click Confirm.
On the O&M > Deployments page, find the target job, click Start in the Actions column, and select Initial Mode.

Step 4: Query the result

In ApsaraDB RDS, run the following statement to view the aggregated and sorted results.

SELECT * FROM `electric_info_sortlistagg`;

The output confirms that the statuses for each user were correctly aggregated and sorted: user_id=1222 corresponds to status_sort=TD,LD,TD,TD; user_id=1333 corresponds to status_sort=TD,LD,LD,LD; and user_id=1555 corresponds to status_sort=TD,LD.

References

For information about the built-in functions that Flink supports, see Built-in functions.
For information about deploying and starting jobs, see Deploy a job and Start a job.
To modify job running parameters, see Configure job deployment. Some parameters can be updated dynamically to reduce downtime caused by stopping and restarting jobs. For more information, see Dynamic scaling and parameter updates.
For information about using Python custom functions in SQL jobs, see Custom functions.