Dynamic Flink CEP — JSON Rule Format

This guide defines the JSON format for expressing Complex Event Processing (CEP) rules used by dynamic Flink CEP. Use it to author, store, and update CEP patterns without changing Java code.

Audience

• Risk control platform developers who integrate or encapsulate CEP rules.
• Risk strategy authors who understand CEP concepts but prefer to write rules in JSON rather than Java.

Overview

A CEP pattern is modeled as a graph:

• A node describes a pattern that matches specific events.
• An edge describes the event selection (contiguity) strategy that moves from one node to the next.
• Graphs can be nested, so a graph may act as a node in a larger graph.

The sections below specify the JSON objects used to describe nodes, quantifiers, conditions, edges, and composite graphs.

Node

A Node represents a complete pattern.

Quantifier

A Quantifier defines how frequently and in what manner a pattern should match.

Times object

Specifies lower/upper bounds and an optional inner window.

1"times": {
2  "from": 3,
3  "to": 3,
4  "windowTime": {
5    "unit": "MINUTES",
6    "size": 12
7  }
8}

• from, to: INTEGER values.
• windowTime: may be null, or an object with:
  • unit: DAYS | HOURS | MINUTES | SECONDS | MILLISECONDS
  • size: numeric (INTEGER or LONG).

Condition

A Condition filters events for a node.

CLASS condition

Use a custom Java class to evaluate the condition.

Example

1{
2  "type": "CLASS",
3  "className": "ververica.cep.demo.StartCondition"
4}

CLASS with custom parameters

Use when you need to update parameters dynamically without recompiling code.

Example

1{
2  "type": "CLASS",
3  "className": "ververica.cep.demo.CustomMiddleCondition",
4  "args": ["A", "B", "C"]
5}

AVIATOR condition

Evaluate an Aviator expression that you can update at runtime.

Example

1{
2  "type": "AVIATOR",
3  "expression": "price > 10"
4}

GROOVY condition

Evaluate a Groovy expression that you can update at runtime.

Example

1{
2  "type": "GROOVY",
3  "expression": "price > 5.0 && name.contains(\"mid\")"
4}

Edge

An Edge connects two nodes and defines the event selection strategy between them.

Valid values for type: STRICT, SKIP_TILL_NEXT, SKIP_TILL_ANY, NOT_FOLLOW, NOT_NEXT. See Contiguity.

GraphNode (extends Node)

A GraphNode is a composite pattern sequence. Each item in nodes is an independent Node; each item in edges specifies how to transition between them.

Additional fields on top of Node:

Graph window

Two window semantics are supported:

• FIRST_AND_LAST: Maximum time between the start and end of a full composite match.
• PREVIOUS_AND_CURRENT: Maximum time between adjacent child pattern matches.

Example

1"window": {
2  "type": "FIRST_AND_LAST",
3  "time": {
4    "unit": "DAYS",
5    "size": 1
6  }
7}

• unit: DAYS | HOURS | MINUTES | SECONDS | MILLISECONDS
• size: numeric (INTEGER or LONG).

AfterMatchSkipStrategy

Controls which partial matches are kept or discarded after a match is emitted.

Valid type values

• NO_SKIP — return every successful match (default).
• SKIP_TO_NEXT — discard partial matches that start with the same event.
• SKIP_PAST_LAST_EVENT — discard partial matches that start between the beginning and end of the emitted match.
• SKIP_TO_FIRST — discard partial matches that start between the beginning of the match and the first occurrence of patternName.
• SKIP_TO_LAST — discard partial matches that start between the beginning of the match and the last occurrence of patternName.

Contiguity (event selection)

Quantifier properties

Examples

Example 1 — Common pattern

Goal (10‑minute window during a promotion):

1. User obtains venue coupons (StartCondition) once — optional.
2. User adds items to cart (MiddleCondition) at least 3 times.
3. User does not complete payment (EndCondition).

Equivalent Java

1Pattern<Event, Event> pattern =
2    Pattern.<Event>begin("start")
3            .where(new StartCondition())
4            .optional()
5            .followedBy("middle")
6            .where(new MiddleCondition())
7            .timesOrMore(3)
8            .notFollowedBy("end")
9            .where(new EndCondition())
10            .within(Time.minutes(10));

JSON

1{
2  "name": "end",
3  "quantifier": {
4    "consumingStrategy": "SKIP_TILL_NEXT",
5    "properties": ["SINGLE"],
6    "times": null,
7    "untilCondition": null
8  },
9  "condition": null,
10  "nodes": [
11    {
12      "name": "end",
13      "quantifier": {
14        "consumingStrategy": "SKIP_TILL_NEXT",
15        "properties": ["SINGLE"],
16        "times": null,
17        "untilCondition": null
18      },
19      "condition": {
20        "className": "ververica.cep.demo.condition.EndCondition",
21        "type": "CLASS"
22      },
23      "type": "ATOMIC"
24    },
25    {
26      "name": "middle",
27      "quantifier": {
28        "consumingStrategy": "SKIP_TILL_NEXT",
29        "properties": ["LOOPING"],
30        "times": {"from": 3, "to": 3, "windowTime": null},
31        "untilCondition": null
32      },
33      "condition": {
34        "className": "ververica.cep.demo.condition.MiddleCondition",
35        "type": "CLASS"
36      },
37      "type": "ATOMIC"
38    },
39    {
40      "name": "start",
41      "quantifier": {
42        "consumingStrategy": "SKIP_TILL_NEXT",
43        "properties": ["SINGLE", "OPTIONAL"],
44        "times": null,
45        "untilCondition": null
46      },
47      "condition": {
48        "className": "ververica.cep.demo.condition.StartCondition",
49        "type": "CLASS"
50      },
51      "type": "ATOMIC"
52    }
53  ],
54  "edges": [
55    {"source": "middle", "target": "end", "type": "NOT_FOLLOW"},
56    {"source": "start", "target": "middle", "type": "SKIP_TILL_NEXT"}
57  ],
58  "window": {
59    "type": "FIRST_AND_LAST",
60    "time": {"unit": "MINUTES", "size": 10}
61  },
62  "afterMatchSkipStrategy": {"type": "NO_SKIP", "patternName": null},
63  "type": "COMPOSITE",
64  "version": 1
65}

Example 2 — Condition with custom parameters

You can tailor marketing actions to customer classes without recompiling. Define a class-based condition and adjust its args list at runtime.

Initial condition

1{
2  "type": "CLASS",
3  "className": "org.apache.flink.cep.pattern.conditions.CustomMiddleCondition",
4  "args": ["A", "B"]
5}

Updated condition

1{
2  "type": "CLASS",
3  "className": "org.apache.flink.cep.pattern.conditions.CustomMiddleCondition",
4  "args": ["A", "B", "C"]
5}