Realtime Compute for Apache Flink は、DataStream ジョブにおいて、ルールが動的に更新される Flink CEP タスクをサポートしています。この記事では、リアルタイムマーケティングのシナリオを例に、ルールを動的に読み込んでアップストリームの Kafka トピックからデータを処理する Flink CEP ジョブを構築する方法を説明します。
ユースケース
Flink CEP は、分散アーキテクチャ、ミリ秒レベルのレイテンシー、強力なルール式により、幅広いアプリケーションに適しています。代表的なシナリオは次の 3 つです：
-
リアルタイムリスクコントロール：Flink CEP を使用して、リスクの高いユーザーを特定できます。たとえば、顧客行動ログを分析し、5 分以内に総額 10,000 を超える送金を 10 回以上行ったユーザーを検出できます。
-
リアルタイムマーケティング：Flink CEP を使用して、マーケティング戦略を最適化できます。たとえば、EC のプロモーション期間中の顧客行動ログを分析することで、10 分以内に 3 点を超える商品をカートに追加したものの購入を完了しなかったユーザーを特定し、ターゲットを絞ったマーケティング調整を行えます。Flink CEP は、リアルタイムマーケティングにおける不正防止のシナリオでも有効です。
-
モノのインターネット (IoT)：Flink CEP を使用して、異常状態を検出し、アラートを送信できます。たとえば、シェアサイクルが指定エリアを離れ、15 分以内に戻らない場合にリスクアラートを発報できます。また、IoT センサーと組み合わせて、生産ラインの異常を検出することも可能です。たとえば、温度センサーが設定したしきい値を超える温度を 3 期間連続で報告した場合、アラートがトリガーされます。
例のウォークスルー
この記事では、動的 CEP を使用してこれらのシナリオに対応する方法を示します。この例では、顧客行動ログを ApsaraMQ for Kafka に保存します。Flink CEP ジョブは、ApsaraDB RDS for MySQL データベース内のルールテーブルをポーリングしながら、このデータを消費します。ポリシー管理者が追加した最新のルールを取得し、それらを使用してイベントをマッチングします。マッチが発生すると、ジョブはアラートを送信するか、関連情報を別のデータストアに書き込みます。次の図は、データパイプライン全体を示しています。
このウォークスルーでは、最初に Flink CEP ジョブを起動し、続いてルール 1 を挿入します。ルール 1 は、action が 0 のイベントが 3 回連続して発生した後、action が 1 ではないイベントが続くシーケンスにマッチします。これは、ユーザーが商品を 3 回閲覧したものの購入しなかったことを示します。
前提条件
-
Realtime Compute for Apache Flink ワークスペースが作成されていること。詳細については、「Realtime Compute for Apache Flink のアクティベート」をご参照ください。
-
RAM ユーザーまたは RAM ロールを使用する場合は、Flink コンソールに必要な権限が付与されている必要があります。詳細については、「権限管理」をご参照ください。
-
アップストリームおよびダウンストリームストレージ：
-
ApsaraDB RDS for MySQL インスタンスが作成されていること。詳細については、「ApsaraDB RDS for MySQL インスタンスの作成」をご参照ください。
-
ApsaraMQ for Kafka インスタンスが作成されていること。詳細については、「ApsaraMQ for Kafka インスタンスの作成」をご参照ください。
-
手順
この記事では、行動ログが特定のルールにマッチするユーザーを監視および記録する Flink CEP ジョブを作成し、動的に更新する方法について説明します。
ステップ 1：テストデータの準備
上流 Kafka トピックの準備
ApsaraMQ for Kafka コンソール にログインします。
-
シミュレートされたユーザー行動ログを保存するために、
demo_topicという名前のトピックを作成します。詳細については、「ステップ 1：トピックの作成」をご参照ください。
RDS データベースの準備
Data Management (DMS) コンソールで、ApsaraDB RDS for MySQL のテストデータを準備します。
-
特権アカウントで ApsaraDB RDS for MySQL インスタンスにログインします。
詳細については、「DMS を使用した ApsaraDB RDS for MySQL インスタンスへのログイン」をご参照ください。
-
Flink CEP ジョブのルールを保存する
rds_demoルールテーブルを作成します。ルールにマッチするデータを保存するmatch_resultsテーブルを作成します。アクティブな [SQLConsole] ウィンドウで、次のコマンドを入力し、[実行] をクリックします。
CREATE DATABASE cep_demo_db; USE cep_demo_db; CREATE TABLE rds_demo ( `id` VARCHAR(64), `version` INT, `pattern` VARCHAR(4096), `function` VARCHAR(512) ); CREATE TABLE match_results ( rule_id INT, rule_version INT, user_id INT, user_name VARCHAR(255), production_id INT, PRIMARY KEY (rule_id,rule_version,user_id,production_id) );rds_demoルールテーブルの各行は単一のルールを表し、異なるルールとそのバージョンを区別するためのidとversion、CEP API の パターン オブジェクトを記述するpatternフィールド、およびパターンにマッチするイベントシーケンスの処理方法を記述するfunctionフィールドが含まれます。match_resultsテーブルの各行は、特定の製品に対するユーザーの行動が特定のルールに適合したマッチを表します。このレコードは、関連製品のクーポン送信などの対応する販売戦略の策定に使用できます。
ステップ 2： IP ホワイトリストの設定
Flink ジョブが ApsaraDB RDS for MySQL インスタンスにアクセスできるようにするには、Realtime Compute for Apache Flink ワークスペースの CIDR ブロックをインスタンスの IP アドレスホワイトリストに追加します。
-
Realtime Compute for Apache Flink ワークスペースの VPC CIDR ブロックを取得します。
-
対象の [ワークスペース] の [操作] 列で、 を選択します。
-
[ワークスペースの詳細] ダイアログボックスで、フルマネージド Flink [vSwitch] の [CIDR ブロック] を確認します。
-
フルマネージド Flink の CIDR ブロックを ApsaraDB RDS for MySQL インスタンスの IP アドレスホワイトリストに追加します。
詳細については、「IP アドレスホワイトリストの設定」をご参照ください。[ホワイトリストの変更] ダイアログボックスで、[ホワイトリスト内のIPアドレス] テキストボックスに Flink の CIDR ブロックを入力します。複数の CIDR ブロックがある場合は、カンマで区切ります。次に、[OK] をクリックします。
ステップ 3：CEP ジョブの開発と起動
この記事のすべてのコードは、GitHub リポジトリで入手できます。デモンストレーションのため、この記事のサンプルコードは timeOrMoreAndWindow ブランチで若干修正されています。完全な ververica-cep-demo-master.zip ファイルをダウンロードして参照できます。
-
ジョブの Maven POM ファイルに、
flink-cepをプロジェクトの依存関係として追加します。その他の Flink 関連 JAR パッケージの処理および競合の解決については、「Flink 環境の依存関係の設定」をご参照ください。
<dependency> <groupId>com.alibaba.ververica</groupId> <artifactId>flink-cep</artifactId> <version>1.17-vvr-8.0.8</version> <scope>provided</scope> </dependency> -
ジョブコードを開発します。
-
Kafka Source を作成します。
コードの記述方法の詳細については、「Kafka DataStream コネクター」をご参照ください。
-
CEP.dynamicPatterns()API を実装します。CEP の動的なルール変更と複数ルールのマッチングをサポートするため、Realtime Compute for Apache Flink は
CEP.dynamicPatterns()API を定義しています。この API は次のように定義されています。public static <T, R> SingleOutputStreamOperator<R> dynamicPatterns( DataStream<T> input, PatternProcessorDiscovererFactory<T> discovererFactory, TimeBehaviour timeBehaviour, TypeInformation<R> outTypeInfo)次の表は、この API のパラメーターを説明しています。実際のユースケースに基づいてパラメーター値を更新できます。
パラメーター
説明
DataStream<T> input入力イベントストリーム。
PatternProcessorDiscovererFactory<T> discovererFactoryPatternProcessorDiscovererを作成するファクトリー。このディスカバラーは、最新のルールを取得し、対応するPatternProcessorインスタンスを構築します。TimeBehaviour timeBehaviourFlink CEP ジョブがイベントの時間属性をどのように処理するかを指定します。有効な値：
-
TimeBehaviour.ProcessingTime：処理時間に基づいてイベントを処理します。 -
TimeBehaviour.EventTime：イベント時間に基づいてイベントを処理します。
TypeInformation<R> outTypeInfo出力ストリームの型情報を指定します。
DataStream、TimeBehaviour、TypeInformation などの一般的な Flink の概念の詳細については、「DataStream API」、「イベント時間と処理時間」、および「TypeInformation」をご参照ください。
PatternProcessorインターフェイスは重要なコンポーネントです。PatternProcessorには、イベントのマッチング方法を記述する特定のPattern、およびアラートの送信などマッチの処理方法を記述するPatternProcessFunctionが含まれます。また、PatternProcessorを識別するためのidとversionも含まれます。詳細については、「提案」をご参照ください。patternProcessorDiscovererFactoryは、最新のPatternProcessorを取得するためのディスカバラーを作成します。サンプルコードには、外部ストアから新しいPatternProcessorインスタンスを定期的にポーリングする方法を示す抽象クラスが含まれています。public abstract class PeriodicPatternProcessorDiscoverer<T> implements PatternProcessorDiscoverer<T> { ... @Override public void discoverPatternProcessorUpdates( PatternProcessorManager<T> patternProcessorManager) { // 定期的にパターンプロセッサーの更新を検出します。 timer.schedule( new TimerTask() { @Override public void run() { if (arePatternProcessorsUpdated()) { List<PatternProcessor<T>> patternProcessors = null; try { patternProcessors = getLatestPatternProcessors(); } catch (Exception e) { e.printStackTrace(); } patternProcessorManager.onPatternProcessorsUpdated(patternProcessors); } } }, 0, intervalMillis); } ... }Realtime Compute for Apache Flink は、ApsaraDB RDS for MySQL や Hologres などの JDBC プロトコルをサポートするデータベースから最新のルールを取得するための
JDBCPeriodicPatternProcessorDiscovererの実装を提供しています。使用する際には、次のパラメーターを指定する必要があります。パラメーター
説明
jdbcUrlデータベースの JDBC URL。
jdbcDriverデータベースドライバーのクラス名。
tableNameデータベーステーブルの名前。
initialPatternProcessorsデータベース内のルールテーブルが空の場合に使用するデフォルトの
PatternProcessorです。intervalMillisデータベースのポーリング間隔 (ミリ秒単位)。
コード内では、次のように使用できます。ジョブは、マッチしたルールを Flink TaskManager に出力します。
// import ...... public class CepDemo { public static void main(String[] args) throws Exception { ...... // DataStream ソース DataStreamSource<Event> source = env.fromSource( kafkaSource, WatermarkStrategy.<Event>forMonotonousTimestamps() .withTimestampAssigner((event, ts) -> event.getEventTime()), "Kafka Source"); env.setParallelism(1); // userId と productionId でキーを設定します。 // 注：同じキーを持つイベントのみが処理され、マッチングが行われます。 KeyedStream<Event, Tuple2<Integer, Integer>> keyedStream = source.assignTimestampsAndWatermarks( WatermarkStrategy.<Event>forGenerator(ctx -> new EventBoundedOutOfOrdernessWatermarks(Duration.ofSeconds(5))) ).keyBy(new KeySelector<Event, Tuple2<Integer, Integer>>() { @Override public Tuple2<Integer, Integer> getKey(Event value) throws Exception { return Tuple2.of(value.getId(), value.getProductionId()); } }); SingleOutputStreamOperator<String> output = CEP.dynamicPatterns( keyedStream, new JDBCPeriodicPatternProcessorDiscovererFactory<>( params.get(JDBC_URL_ARG), JDBC_DRIVE, params.get(TABLE_NAME_ARG), null, Long.parseLong(params.get(JDBC_INTERVAL_MILLIS_ARG))), Boolean.parseBoolean(params.get(USING_EVENT_TIME)) ? TimeBehaviour.EventTime : TimeBehaviour.ProcessingTime, TypeInformation.of(new TypeHint<String>() {})); output.print(); // ジョブをコンパイルして実行します env.execute("CEPDemo"); } }説明デモンストレーションのため、デモコードは入力データストリームを
idとproductionIdをキーとして設定してからCEP.dynamicPatterns()に接続しています。これは、同じidとproductionIdを持つイベントのみがルールマッチングの対象となることを意味します。異なるキーを持つイベント同士はマッチングされません。 -
-
-
Realtime Compute for Apache Flink コンソールで、JAR ファイルをアップロードし、JAR ジョブをデプロイします。詳細については、「ジョブのデプロイ」をご参照ください。
すぐに開始できるように、テスト用の cep-demo.jar ファイルをダウンロードできます。次の表は、デプロイ時に設定するパラメーターを説明しています。
説明上流の Kafka ソースが空で、データベースのルールテーブルにデータが含まれていないため、ジョブは起動後に出力を生成しません。
パラメーター
説明
[デプロイモード]
[Stream Mode] を選択します。
[デプロイメント名]
JAR ジョブの名前を入力します。
[エンジンバージョン]
エンジンバージョンの詳細については、「エンジンバージョン」および「ライフサイクルポリシー」をご参照ください。推奨バージョンまたは安定バージョンのご使用を推奨します。バージョンタグは次のとおりです。
-
推奨バージョン：現在のメジャーバージョンの最新マイナーバージョン。
-
安定バージョン：製品サポート期間内にあり、過去の不具合が修正されたメジャーバージョンの最新マイナーバージョン。
-
通常バージョン：製品サポート期間内にあるその他のマイナーバージョン。
-
EOS バージョン：製品サポート期間を超えたバージョン。
[JAR URL]
パッケージ化された JAR ファイルをアップロードするか、提供されているテスト JAR ファイルをアップロードします。
[エントリポイントクラス]
com.alibaba.ververica.cep.demo.CepDemoを入力します。[エントリポイントのメイン引数]
独自に開発したジョブを使用していて、上流および下流のストレージ情報をすでに設定している場合は、このフィールドを空白のままにできます。ただし、提供されているテスト JAR を使用している場合は、このパラメーターを設定する必要があります。コードは次のとおりです。
--kafkaBrokers YOUR_KAFKA_BROKERS --inputTopic YOUR_KAFKA_TOPIC --inputTopicGroup YOUR_KAFKA_TOPIC_GROUP --jdbcUrl jdbc:mysql://YOUR_DB_URL:port/DATABASE_NAME?user=YOUR_USERNAME&password=YOUR_PASSWORD --tableName YOUR_TABLE_NAME --jdbcIntervalMs 3000 --usingEventTime false各パラメーターの説明は次のとおりです。
-
kafkaBrokers：Kafka ブローカーアドレス。 -
inputTopic：Kafka トピック名。 -
inputTopicGroup：Kafka コンシューマーグループ。 -
jdbcUrl：データベースの JDBC URL。説明この例の JDBC URL で指定するユーザー名とパスワードは、標準アカウントのものである必要があります。また、パスワードに含めることができるのは英数字のみです。実際の環境では、必要に応じてジョブで異なる認証方法を使用できます。
-
tableName：ターゲットテーブルの名前。 -
jdbcIntervalMs：データベースのポーリング間隔。 -
usingEventTime：処理にイベント時間を使用するかどうかを指定します (true/false)。
説明-
プレースホルダーを、実際の上流および下流ストレージの情報に置き換えてください。
-
本番環境では、プレーンテキストのパスワードの使用は避けてください。変数管理機能を使用することを推奨します。詳細については、「変数管理」をご参照ください。
-
-
[デプロイメントの詳細] ページの [その他の設定] セクションで、次のジョブのランタイムパラメーターを追加します。
実際のアプリケーションでは、
flink-cepJAR はシステムクラスローダーによってロードされ、aviator関連のクラスは通常ユーザー JAR にパッケージ化され、ユーザークラスローダーによってロードされます。以下の 2 つの設定を使用することで、システムクラスローダーがクラスをロードしようとする際にユーザー JAR 内のクラスにアクセスできるようになり、クラスロードの失敗を回避できます。kubernetes.application-mode.classpath.include-user-jar: 'true' classloader.resolve-order: parent-firstランタイムパラメーターの設定の詳細については、「ランタイムパラメーターの設定」をご参照ください。
-
ページで、対象のデプロイメントを見つけ、[アクション] 列の [起動] をクリックします。
ジョブ起動パラメーターの設定の詳細については、「ジョブの起動」をご参照ください。
ステップ 4：ルールの挿入
Flink CEP ジョブの実行中に、ルール 1 を挿入します。このルールは、「action が 0 である 3 つの連続したイベントの後、次のイベントの action が 1 ではない」というものです。これは、ユーザーが商品を 3 回閲覧したものの購入しなかったことを意味します。
-
ApsaraDB RDS for MySQL コンソールにログインします。
-
動的更新ルールを挿入します。
id、version、関数クラス名を含む JSON 文字列を連結し、RDS に挿入します。INSERT INTO rds_demo ( `id`, `version`, `pattern`, `function` ) values( '1', 1, '{"name":"end","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":null,"nodes":[{"name":"end","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":{"className":"com.alibaba.ververica.cep.demo.condition.EndCondition","type":"CLASS"},"type":"ATOMIC"},{"name":"start","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["LOOPING"],"times":{"from":3,"to":3,"windowTime":null},"untilCondition":null},"condition":{"expression":"action == 0","type":"AVIATOR"},"type":"ATOMIC"}],"edges":[{"source":"start","target":"end","type":"SKIP_TILL_NEXT"}],"window":null,"afterMatchStrategy":{"type":"SKIP_PAST_LAST_EVENT","patternName":null},"type":"COMPOSITE","version":1}', 'com.alibaba.ververica.cep.demo.dynamic.DemoPatternProcessFunction') ;データベース内の
patternフィールドの使いやすさと可読性を向上させるため、Realtime Compute for Apache Flink は JSON ベースのルール形式を定義しています。詳細については、「動的 CEP におけるルールの JSON 形式」をご参照ください。上記の SQL ステートメントのpatternフィールドには、シリアル化された JSON 文字列が含まれています。この文字列は、「actionが 0 である 3 つの連続したイベントの後、次のイベントの action が 1 ではない」というシーケンスにマッチするパターンを表しています。説明EndConditionコードでは、条件はaction != 1と定義されています。-
対応する CEP API の説明は次のとおりです。
Pattern<Event, Event> pattern = Pattern.<Event>begin("start", AfterMatchSkipStrategy.skipPastLastEvent()) .where(new StartCondition("action == 0")) .times(3) .followedBy("end") .where(new EndCondition()); -
CepJsonUtilsのメソッドを使用して、対応する JSON 文字列に変換できます。public void printTestPattern(Pattern<?, ?> pattern) throws JsonProcessingException { System.out.println(CepJsonUtils.convertPatternToJSONString(pattern)); } -
対応する JSON 文字列は次のとおりです。
-
-
Kafka クライアントを使用して、
demo_topicトピックにメッセージを送信します。このデモでは、ApsaraMQ for Kafka が提供する [メッセージの送受信を開始] ページを使用してテストメッセージを送信することもできます。
1,田中一郎,0,1,1662022777000 1,田中一郎,0,1,1662022778000 1,田中一郎,0,1,1662022779000 1,田中一郎,0,1,1662022780000[コンソール] 送信方式を選択し、[メッセージキー] フィールドに
1を入力し、[メッセージ内容] フィールドにテストデータを貼り付け、[指定パーティションに送信] を [いいえ] に設定し、メッセージを送信します。ページに 「メッセージが正常に送信されました」 という通知が表示されます。次の表は、
demo_topicのフィールドを説明しています。パラメーター
説明
id
ユーザー ID。
username
ユーザー名。
action
ユーザーアクション。有効な値：
-
0：閲覧操作
-
1：購入アクション
product_id
商品 ID。
event_time
行動イベントが発生した時刻。
-
-
JobManager ログで最新のルールを確認し、TaskManager ログでマッチを確認します。
-
JobManager ログで、
JDBCPeriodicPatternProcessorDiscovererを検索して最新のルールを確認します。ログを見つけるには、[ログ] > [JobManager] に移動し、[ログ] タブをクリックします。検索ボックスにキーワードを入力して、関連するログエントリを見つけます。ログメッセージ
PatternProcessors have been updatedにより、ルールが正常に更新されたことが確認できます。 -
.out で終わる TaskManager ログファイルで、
A match for Pattern of (id, version): (1, 1)を検索して、一致した結果を確認します。デプロイメントの詳細ページで、[ログ] タブをクリックし、[実行中のタスクマネージャー] を選択し、対応する TaskManager の [ログ] サブタブを開きます。[flink.out] ファイルでキーワードを検索して、マッチしたイベントシーケンスを見つけます。
-
-
SELECT * FROM `match_results`;を実行してmatch_resultsテーブルをクエリし、ルールにマッチする結果を確認します。クエリは、
rule_id、rule_version、user_id、user_name、product_idのフィールドを持つ 1 つのレコードを返し、それぞれの値は 1、1、1、田中一郎、1 です。
ステップ 5: マッチングルールの更新
マーケティング戦略には時間的な制約があることがよくあります。このステップでは、3 つの action = 0 イベントが 15 分以内に発生するよう、ルールを更新します。
-
usingEventTimeパラメーターをtrueに設定します。-
ページで、目的のデプロイを見つけ、[操作]列の[キャンセル]をクリックします。
-
で [編集] をクリックして
usingEventTimeパラメーターをtrueに設定し、 [保存] をクリックします。 -
ジョブを再度[開始]します。
-
-
新しいルールを挿入します。
対応する CEP API の説明は次のとおりです：
Pattern<Event, Event> pattern = Pattern.<Event>begin("start", AfterMatchSkipStrategy.skipPastLastEvent()) .where(new StartCondition("action == 0")) .timesOrMore(3,Time.minutes(15)) .followedBy("end") .where(new EndCondition()); printTestPattern(pattern);新しいルールを
rds_demoテーブルに挿入します。# このデモでのルールの競合を避けるため、まず以前のルールを削除します。 DELETE FROM `rds_demo` WHERE `id` = 1; # 新しいルール (バージョン 2) を挿入します。15 分以内に 3 回連続で `action = 0` のイベントが発生した後に、action が 1 ではないイベントが続く場合にマッチします。 INSERT INTO rds_demo (`id`,`version`,`pattern`,`function`) values('1',2,'{"name":"end","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":null,"nodes":[{"name":"end","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":{"className":"com.alibaba.ververica.cep.demo.condition.EndCondition","type":"CLASS"},"type":"ATOMIC"},{"name":"start","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["LOOPING"],"times":{"from":3,"to":3,"windowTime":{"unit":"MINUTES","size":15}},"untilCondition":null},"condition":{"expression":"action == 0","type":"AVIATOR"},"type":"ATOMIC"}],"edges":[{"source":"start","target":"end","type":"SKIP_TILL_NEXT"}],"window":null,"afterMatchStrategy":{"type":"SKIP_PAST_LAST_EVENT","patternName":null},"type":"COMPOSITE","version":2}','com.alibaba.ververica.cep.demo.dynamic.DemoPatternProcessFunction'); -
Kafka コンソールで、マッチを発生させるために 8 つのメッセージを送信します。
次は 8 つのサンプルメッセージです：
2,Tom,0,1,1739584800000 #10:00 2,Tom,0,1,1739585400000 #10:10 2,Tom,0,1,1739585700000 #10:15 2,Tom,0,1,1739586000000 #10:20 3,Ali,0,1,1739586600000 #10:30 3,Ali,0,1,1739588400000 #11:00 3,Ali,0,1,1739589000000 #11:10 3,Ali,0,1,1739590200000 #11:30 -
SELECT * FROM `match_results`;を実行し、match_resultsテーブルでルールにマッチする結果を確認します。クエリによって 1 件のレコードが返されます。このレコードは
user_nameが Tom、rule_versionが 2 となっており、新しいルールにマッチしたことを示しています。この結果は、Tom の行動のみが新しいルールにマッチすることを示しています。これは、Ali のアクションが 15 分以上にわたって発生したためです。期間限定のプロモーションでは、特定の時間枠内で商品を繰り返し訪問するユーザーにクーポンを送信し、購入を促すことができます。