すべてのプロダクト
Search

このプロダクト

    Realtime Compute for Apache Flink:UDAF を使用したデータのソートと集計

    ドキュメントセンター

    Realtime Compute for Apache Flink:UDAF を使用したデータのソートと集計

    最終更新日:Jun 21, 2026

    このトピックでは、住宅用電力網の端末データを使用して、Realtime Compute for Apache Flink コンソールでユーザー定義集計関数 (UDAF) を使用してデータをマージおよびソートする方法を説明します。

    サンプルデータ

    electric_info テーブルには、住宅用電力網の端末データが含まれています。このテーブルには、イベント ID (event_id)、ユーザー ID (user_id)、イベント時間 (event_time)、および端末ステータス (status) が含まれます。各ユーザーの status 値を集計し、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

    • 期待される結果

      user_id

      status

      1222

      TD,LD,TD,TD

      1333

      TD,LD,LD,LD

      1555

      TD,LD

    ステップ1：データソースの準備

    この例では、データソースとして ApsaraDB RDS を使用します。

    1. ApsaraDB RDS for MySQL インスタンスを作成します。

      説明

      ApsaraDB RDS for MySQL インスタンスは、Realtime Compute for Apache Flink ワークスペースと同じ VPC にある必要があります。異なる VPC にある場合は、「ネットワーク接続」をご参照ください。

    2. データベースとアカウントを作成します。

      electric という名前のデータベースと、このデータベースに対する読み取りおよび書き込み権限を持つアカウントを作成します。

    3. Data Management (DMS) を使用して ApsaraDB RDS for MySQL インスタンスにログインし、electric データベースに electric_info テーブルと electric_info_SortListAgg テーブルを作成して、データを挿入します。

      CREATE TABLE `electric_info` (
  event_id bigint NOT NULL PRIMARY KEY COMMENT 'イベント ID',
  user_id bigint NOT NULL COMMENT 'ユーザー ID', 
  event_time timestamp NOT NULL COMMENT 'イベント時間',
  status varchar(10) NOT NULL COMMENT 'ユーザー端末のステータス'
);
CREATE TABLE `electric_info_SortListAgg` (
  user_id bigint NOT NULL PRIMARY KEY COMMENT 'ユーザー ID', 
  status_sort varchar(50) NULL COMMENT 'イベント時間で昇順にソートされたユーザー端末のステータス'
);
-- データを準備
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');

    ステップ2：UDAF の登録

    1. ASI_UDX-1.0-SNAPSHOT.jar パッケージをダウンロードします。

      pom.xml ファイルは、Flink バージョン 1.17.1 のこのカスタム関数に必要な最小限の依存関係で設定されています。カスタム関数の詳細については、「カスタム関数」をご参照ください。

    2. サンプルコード ASI_UDAF は、複数の行を 1 つの行にマージし、指定された列でデータをソートします。ビジネスニーズに合わせてコードを変更できます。

      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{
	/**アキュムレータクラス*/
	public static class AcList {
		public  List<String> list;
	}
	/**集計関数クラス*/
	public static class SortListAgg extends AggregateFunction<String,AcList> {
		public String getValue(AcList asc) {
			/**特定のルールに基づいてリスト内のデータをソートします*/
			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]);
				}
			});
			/**ソートされたリストを走査し、必要なフィールドを抽出して文字列に結合します*/
			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;
		}
		/**アキュムレータを作成するメソッド*/
		public AcList createAccumulator() {
			AcList ac = new AcList();
			List<String> list = new ArrayList<String>();
			ac.list = list;
			return ac;
		}
		/**アキュムレーションメソッド：入力データをアキュムレータに追加します*/
		public void accumulate(AcList acc, String tuple1) {
			acc.list.add(tuple1);
		}
		/**リトラクションメソッド*/
		public void retract(AcList acc, String num) {
		}
	}
}

    3. UDAF を登録します。

      UDAF を登録すると、そのコードを他のジョブで再利用できます。Java UDAF の場合、JAR を依存関係ファイルとしてアップロードすることもできます。詳細については、「ユーザー定義集計関数 (UDAF)」をご参照ください。

      1. Realtime Compute for Apache Flink コンソールにログインします。

      2. 目的のワークスペースを見つけ、[アクション] 列の [コンソール] をクリックします。

      3. 左側のナビゲーションペインで、[開発] > [ETL] を選択します。

      4. [機能] タブで、[UDF の登録] をクリックします。

    4. [ファイルを選択] セクションで、ステップ 1 の JAR ファイルをアップロードし、[OK] をクリックします。

      ダイアログボックスには、[Upload File] と [External URL] の 2 つの登録方法があります。[UDF Name] を指定する必要があり、任意で [dependency file] をアップロードできます。

      説明

      • UDF の JAR ファイルは、ワークスペースに関連付けられている OSS バケットの sql-artifacts ディレクトリにアップロードされます。

      • Realtime Compute for Apache Flink コンソールは UDF JAR ファイルを解析し、Flink UDF、UDAF、UDTF インターフェイスを使用するクラスを検出します。クラス名を自動的に抽出し、[Function Name] フィールドに設定します。

    5. [関数の管理] ダイアログボックスで、[関数を作成] をクリックします。

      登録された UDF は、SQL エディターページの左側にある[関数] リストに表示されます。

    ステップ3：Flink ジョブの作成

    1. [開発] > [ETL] ページで、[新規作成] をクリックします。

    2. [空白のストリームドラフト] をクリックします。

    3. [次へ] をクリックします。

    4. [新規ドラフト] ダイアログボックスで、ジョブ設定を設定します。

      パラメーター

      説明

      [ファイル名]

      ジョブの一意の名前。

      説明

      ジョブ名は現在のプロジェクト内で一意である必要があります。

      [保存場所]

      ジョブの保存場所。

      既存のフォルダーの横にある 新建文件夹 アイコンをクリックして、サブフォルダーを作成することもできます。

      [エンジンバージョン]

      ジョブの Flink エンジンバージョン。これは、pom.xml ファイルで指定されたバージョンと一致する必要があります。

      エンジンバージョン、バージョンのマッピング、ライフサイクル情報の詳細については、「エンジンバージョン」をご参照ください。

    5. DDL および DML ステートメントを記述します。

      -- 一時テーブル 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'
);
-- electric_info テーブルからデータを集計し、electric_info_sortlistagg テーブルに挿入します。
-- 登録されたカスタム関数 ASI_UDAF$SortListAgg のパラメーターとして、status と event_time を連結した文字列を渡します。
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;

      次の表でパラメーターについて説明します。実際のニーズに基づいて変更してください。MySQL コネクタのパラメーターの詳細については、「MySQL コネクタ」をご参照ください。

      パラメーター

      説明

      注意

      connector

      コネクタのタイプ。

      この例では、値は mysql に固定されています。

      hostname

      MySQL データベースの IP アドレスまたはホスト名。

      この例では、ApsaraDB RDS for MySQL インスタンスの内部エンドポイントを使用します。

      username

      MySQL データベースサービスのユーザー名。

      なし。

      password

      MySQL データベースサービスのパスワード。

      この例では、セキュリティリスクを回避するために、パスワードには変数の mysql_pw を使用します。詳細については、「変数」をご参照ください。

      database-name

      MySQL データベースの名前。

      この例では、「ステップ1：データソースの準備」で作成したデータベース electric を使用します。

      table-name

      MySQL テーブルの名前。

      この例では、electric_info または electric_info_sortlistagg に設定します。

      port

      MySQL データベースサービスのポート番号。

      なし。

    6. (任意) 右上隅で、[検証] と [デバッグ] をクリックします。これらの機能の詳細については、「ジョブ開発の概要」をご参照ください。

    7. [デプロイ] をクリックし、次に [確認] をクリックします。

    8. O&M > デプロイメント ページで、対象のジョブを見つけ、[操作] 列の [開始] をクリックし、[初期モード] を選択します。

    ステップ4：結果のクエリ

    ApsaraDB RDS で、次のステートメントを実行して、集計およびソートされた結果を表示します。

    SELECT * FROM `electric_info_sortlistagg`;

    出力は、各ユーザーのステータスが正しく集計およびソートされたことを示しています：user_id=1222 は status_sort=TD,LD,TD,TD に対応し、user_id=1333 は status_sort=TD,LD,LD,LD に対応し、user_id=1555 は status_sort=TD,LD に対応します。

    関連ドキュメント