Building a Real‑Time Risk Control Engine with Flink 2.2.1, CEP, and Aviator

Project Overview

Real‑time risk control engine for e‑commerce and financial scenarios. Built on Apache Flink 2.2.1 with CEP for millisecond‑level anomaly detection, uses Aviator for dynamic rule evaluation, and pushes results to Redis, MySQL, and Kafka for interception, alerting, and audit decisions.

Core Capabilities

Real‑time event ingestion – three Kafka topics (login, order, activity) are consumed via Flink 2.x KafkaSource with watermark support and event‑time semantics.

CEP pattern matching – six built‑in risk patterns (login‑failure storm, remote login, frequent order, rapid order, activity brush, duplicate coupon) defined with Flink CEP Pattern API using followedBy and within for flexible continuous matching and time windows.

Dynamic rule engine – Aviator 5.4.3 expression engine stores rules in a MySQL table; a custom RuleSourceFunction pulls the table every 30 seconds and broadcasts updates via BroadcastState, enabling hot rule updates without restarting the job.

Multi‑channel result output – risk decisions are written simultaneously to Redis (TTL 1 h cache), MySQL (persistent audit) and Kafka (downstream consumption).

Technical Highlights

Flink 2.2.1 with EventTime + Watermark semantics.

Flink CEP Pattern API with followedBy and within for time‑windowed matching.

BroadcastState + RuleSourceFunction for rule hot‑updates.

Aviator 5.4.3 with compiled cache, regex support and string functions.

Kafka 4.1.x in KRaft mode (no Zookeeper) using cp‑kafka:8.1.4.

Redis 8.8 accessed via Jedis 5.2.0.

Flink checkpointing guarantees exactly‑once semantics; Kafka offsets are committed automatically.

Applicable Scenarios

E‑commerce marketing risk: coupon abuse, order‑brush, activity‑brush detection.

Account security: brute‑force login, remote login alerts, credential‑stuffing detection.

Payment risk: high‑frequency large‑amount orders, suspicious payment channels, ultra‑fast transaction interception.

System Architecture

Technology Stack

Stream Processing: Apache Flink 2.2.1 (EventTime + Watermark)

CEP: Flink CEP 2.2.1

Rule Engine: Aviator 5.4.3

Message Queue: Kafka (KRaft mode) cp‑kafka:8.1.4

Cache: Redis 8.8 (Jedis 5.2.0)

Database: MySQL 8.0

Web Backend: Spring Boot 3.2.0

ORM: MyBatis Plus 3.5.5

Frontend: React 18.3 + Ant Design 5.22

Build: Maven 3.8+ + Vite 5.4

Quick Start

Start Docker dependencies:

# method 1: docker-compose
cd && docker compose up -d
# method 2: script (auto creates topics and initializes DB)
cd scripts && ./docker-start.sh start

Start the web management backend:

cd flink-risk-web
mvn spring-boot:run

Start the frontend UI:

cd frontend
npm install
npm run dev

Compile and run the Flink job:

# compile package
cd flink-risk-job
mvn package -DskipTests
# run locally (IDEA debug or command line)
java -cp target/classes:$(mvn dependency:build-classpath -Dmdep.outputFile=/dev/stdout) com.qinyadan.risk.RiskControlApplication

Send simulated data for testing:

cd scripts
./event-simulator.sh login_failure user_001   # login‑failure scenario
./event-simulator.sh order_frequent user_002   # frequent order scenario
./event-simulator.sh activity_frequent user_003   # frequent activity scenario
./event-simulator.sh coupon_repeat user_004   # duplicate coupon scenario
./event-simulator.sh all user_001   # all scenarios
./event-simulator.sh stress   # stress test

CEP Risk Scenarios

Login failure storm – three consecutive events with success=false within 5 s; rule failCount >= 3 → HIGH/BLOCK.

Remote login – two consecutive successful logins from different locations within 5 s; rule patternType == "LOCATION_CHANGE" → MEDIUM/VERIFY.

Frequent order – three orders within 5 s; rule patternType == "ORDER_FREQUENT" → HIGH/BLOCK.

Rapid order – two orders within 5 s; rule patternType == "ORDER_RAPID" → CRITICAL/BLOCK.

Frequent activity participation – four activity events within 5 s; rule participationCount >= 4 → MEDIUM/WARN.

Duplicate coupon claim – three coupon‑claim events within 5 s; rule patternType == "COUPON_REPEAT" → HIGH/BLOCK.

Rule Engine Details

Rules are stored in MySQL table rule_config. The Flink job uses a custom RuleSourceFunction that pulls the table every 30 seconds and broadcasts the rule set via BroadcastStream to all parallel task instances, enabling hot updates without job restart.

Aviator expressions can reference CEP context variables such as eventType, patternType, userId, failCount, amount, orderCount, participationCount, actionType, channel, and success. The system ships with 15 pre‑defined rules covering login (4), order (5), activity (4) and generic (2) scenarios.

Kafka Topic Formats

Login events (JSON):

{
  "userId":"user_001",
  "eventId":"login_xxx",
  "eventType":"LOGIN",
  "timestamp":1710000000000,
  "ip":"192.168.1.1",
  "deviceId":"dev_001",
  "success":false,
  "failureReason":"密码错误",
  "location":"北京",
  "browser":"Chrome"
}

Order events (JSON):

{
  "userId":"user_001",
  "eventId":"order_xxx",
  "eventType":"ORDER",
  "timestamp":1710000000000,
  "orderId":"ORD001",
  "amount":50000.00,
  "productId":"SKU001",
  "productName":"异常商品",
  "quantity":30,
  "paymentMethod":"可疑支付",
  "deliveryAddress":"异常地址",
  "processTime":0
}

Activity events (JSON):

{
  "userId":"user_001",
  "eventId":"act_xxx",
  "eventType":"ACTIVITY",
  "timestamp":1710000000000,
  "activityId":"ACT001",
  "activityName":"双11大促",
  "couponCode":"CPN001",
  "couponValue":50.00,
  "actionType":"PARTICIPATE",
  "participationCount":5,
  "channel":"app",
  "source":"push"
}

Utility Scripts

docker-start.sh

controls Docker containers (start, stop, restart, status). event-simulator.sh generates test events for each scenario, batch runs, stress tests, and topic cleanup.

Source Code

Repository: https://github.com/liuzm/flink-risk-control