All Products
Search

This Product

    Realtime Compute for Apache Flink:E-commerce alerting with dynamic Flink CEP

    Document Center

    Realtime Compute for Apache Flink:E-commerce alerting with dynamic Flink CEP

    Last Updated:Apr 13, 2026

    Flink's Complex Event Processing (CEP) library lets you dynamically process complex event streams, detect specific event patterns in real time, and trigger alerts. In e-commerce marketing, you can use Flink CEP to monitor user behavior and transaction data in real time, identify abnormal or key events, and send timely alerts.

    Background

    As the e-commerce industry expands, the volume of user behavior and transaction data is increasing exponentially. Traditional batch processing methods are too slow to identify and respond to abnormal behavior, system risks, and user churn. In contrast, a dynamic Complex Event Processing (CEP) engine can model and analyze multi-stage user behavior to automatically detect complex event patterns and trigger alerts at the first sign of risk. This is the core advantage of dynamic CEP in real-time business operations. It has the following three key features:

    • Low latency: Enables millisecond-level responses and supports in-process alerting rather than post-event analysis for rapid decision-making.

    • Flexible and configurable rules: Supports dynamic rule updates to adapt quickly to business changes without service restarts.

    • Powerful complex event detection: Supports advanced logic matching, such as multi-event sequences, time windows, and combined conditions, to accurately capture complex business scenarios.

    In the e-commerce industry, typical use cases for dynamic CEP include:

    Scenario

    Description

    Cross-selling and up-selling opportunities

    When browsing products, users often show interest across categories, such as viewing a phone and then looking at headphones or power banks. This behavior presents cross-selling and up-selling opportunities. By precisely recommending complementary products (like phone cases or headphones) or offering bundled deals (such as "Phone + Headphones Combo Discount"), the platform can increase the purchase rate of additional items, raise the average order value, and improve the user experience. This enhances user stickiness and drives business growth.

    High-value product purchase conversion

    A high-value conversion occurs when a user adds a high-value product to the cart and completes the purchase. By identifying this conversion path in real time, the platform can analyze user decision-making characteristics, optimize product recommendation strategies, and increase the repurchase rate and average order value for high-value products.

    High-intent user identification

    A user who browses the same product multiple times in a short period indicates high purchase intent. By identifying this behavior and triggering personalized marketing (such as exclusive coupons or stock reminders), the platform can accelerate the user's decision-making process, improve conversion rates, and enhance the user experience, thereby driving sales growth.

    Price-sensitive user engagement

    Price-sensitive users often repeatedly browse a product and only add it to the cart when the price drops. By analyzing this behavior, the platform can send notifications or targeted offers (such as "The item you're watching is now on sale!") when the price changes. This improves conversion rates and optimizes user engagement efficiency.

    Churn risk alert

    A user who frequently browses products but does not make a purchase for an extended period may be at risk of churning. By identifying such behavior and taking recovery actions (such as sending exclusive coupons or recommending popular products), the platform can effectively reduce the churn rate, extend the user's lifecycle, and increase both user retention and platform revenue.

    Solution architecture

    Flink CEP is a library in Apache Flink for processing complex event patterns. By defining complex event patterns, Flink CEP monitors an event stream in real time, identifies matching event sequences, and outputs the results. The solution architecture can be summarized as follows:

    2

    1. Event Stream

      An event stream is the input source for CEP. It is typically a continuous data stream that contains a series of time-ordered events. Each event can have multiple attributes for subsequent pattern matching.

    2. Pattern and Rule Definitions

      You define event patterns and rules that describe the desired event sequences or combinations. Patterns can include event order, time constraints, and filter conditions. For example, you can define a pattern for "Event A is followed by Event B within 10 seconds."

    3. CEP Engine Analysis

      The CEP engine receives the event stream and analyzes it based on the defined patterns and rules. The engine continuously monitors the event stream and matches incoming events against the defined patterns. During the matching process, the engine considers event order, attribute conditions, and time window constraints.

    4. CEP Matching Outputs

      When an event sequence in the event stream matches a defined pattern, the CEP engine generates an output. This output can be the matched event sequence, an action triggered by a rule, or another user-defined output format. The matching results can be used for subsequent processing, such as alerting, decision-making, or storage.

    Prerequisites

    Step 1: Prepare the environment

    Create an RDS for MySQL instance and data source

    1. Create an ApsaraDB RDS for MySQL database. For more information, see Create a database.

      Create a database named ecommerce for the target instance.

    2. Prepare the MySQL CDC data source.

      1. On the instance details page, click Log On To Database at the top.

      2. In the DMS page that appears, enter the username and password for the database account, and then click Log On.

      3. After you log on, double-click the ecommerce database in the left-side pane to switch to it.

      4. In the SQL Console, enter the following DDL and INSERT statements.

        -- Create rule table 1
CREATE TABLE rds_demo1 (
  `id` VARCHAR(64),
  `version` INT,
  `pattern` VARCHAR(4096),
  `function` VARCHAR(512)
);

-- Create rule table 2
CREATE TABLE rds_demo2 (
  `id` VARCHAR(64),
  `version` INT,
  `pattern` VARCHAR(4096),
  `function` VARCHAR(512)
);

-- Create rule table 3
CREATE TABLE rds_demo3 (
  `id` VARCHAR(64),
  `version` INT,
  `pattern` VARCHAR(4096),
  `function` VARCHAR(512)
);


-- Create rule table 4
CREATE TABLE rds_demo4 (
  `id` VARCHAR(64),
  `version` INT,
  `pattern` VARCHAR(4096),
  `function` VARCHAR(512)
);

-- Create rule table 5
CREATE TABLE rds_demo5 (
  `id` VARCHAR(64),
  `version` INT,
  `pattern` VARCHAR(4096),
  `function` VARCHAR(512)
);


-- Create source tables
CREATE TABLE `click_stream1` (
  id bigint not null primary key auto_increment,  -- Auto-incrementing primary key
  eventTime timestamp,  
  eventType varchar(50),
  productId varchar(50), 
  categoryId varchar(50),  
  categoryCode varchar(80),
  brand varchar(50),
  price decimal(10, 2),
  userId varchar(50), 
  userSession varchar(50)
);

CREATE TABLE `click_stream2` (
  id bigint not null primary key auto_increment,  -- Auto-incrementing primary key
  eventTime timestamp,  
  eventType varchar(50),
  productId varchar(50), 
  categoryId varchar(50),  
  categoryCode varchar(80),
  brand varchar(50),
  price decimal(10, 2),
  userId varchar(50), 
  userSession varchar(50)
);

CREATE TABLE `click_stream3` (
  id bigint not null primary key auto_increment,  -- Auto-incrementing primary key
  eventTime timestamp,  
  eventType varchar(50),
  productId varchar(50), 
  categoryId varchar(50),  
  categoryCode varchar(80),
  brand varchar(50),
  price decimal(10, 2),
  userId varchar(50), 
  userSession varchar(50)
);

CREATE TABLE `click_stream4` (
  id bigint not null primary key auto_increment,  -- Auto-incrementing primary key
  eventTime timestamp,  
  eventType varchar(50),
  productId varchar(50), 
  categoryId varchar(50),  
  categoryCode varchar(80),
  brand varchar(50),
  price decimal(10, 2),
  userId varchar(50), 
  userSession varchar(50)
);


CREATE TABLE `click_stream5` (
  id bigint not null primary key auto_increment,  -- Auto-incrementing primary key
  eventTime timestamp,  
  eventType varchar(50),
  productId varchar(50), 
  categoryId varchar(50),  
  categoryCode varchar(80),
  brand varchar(50),
  price decimal(10, 2),
  userId varchar(50), 
  userSession varchar(50)
);

      5. Click Execute and then click Execute Directly.

    Create Kafka topics and a consumer group

    Create the following Kafka resources by following the instructions in Create resources:

    • Group: clickstream.consumer.

    • Topics: click_stream1, click_stream2, click_stream3, click_stream4, and click_stream5.

      When you create the topics, we recommend that you set the number of partitions to 1. Otherwise, the sample data may not match the expected results in some scenarios.

      image

    Step 2: Synchronize data from MySQL to Kafka

    Synchronizing user clickstream events from MySQL to Kafka reduces the load from multiple jobs on the database.

    1. Create a MySQL catalog. For more information, see Create a MySQL catalog.

      In this example, the catalog is named mysql-catalog, and the default database is ecommerce.

    2. Create a Kafka catalog. For more information, see Manage Kafka JSON catalogs.

      In this example, the catalog is named kafka-catalog.

    3. On the Data Development > ETL page, create a new SQL stream job and copy the following code into the SQL editor.

      CREATE TEMPORARY TABLE `clickstream1` (
  `key_id` BIGINT,
  `value_eventTime` BIGINT,  
  `value_eventType` STRING,
  `value_productId` STRING,
  `value_categoryId` STRING,
  `value_categoryCode` STRING,
  `value_brand` STRING,
  `value_price` DECIMAL(10, 2),
  `value_userId` STRING,
  `value_userSession` STRING,
  -- Define the primary key.
  PRIMARY KEY (`key_id`) NOT ENFORCED,
  ts AS TO_TIMESTAMP_LTZ(value_eventTime, 3),
  WATERMARK FOR ts AS ts - INTERVAL '2' SECOND  -- Define a watermark.
) WITH (
  'connector'='upsert-kafka',
  'topic' = 'click_stream1',
  'properties.bootstrap.servers' = 'alikafka-pre-cn-w******02-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-w******02-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-w******02-3-vpc.alikafka.aliyuncs.com:9092',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'value_',
  'value.fields-include' = 'EXCEPT_KEY'
);


CREATE TEMPORARY TABLE `clickstream2` (
  `key_id` BIGINT,
  `value_eventTime` BIGINT,  
  `value_eventType` STRING,
  `value_productId` STRING,
  `value_categoryId` STRING,
  `value_categoryCode` STRING,
  `value_brand` STRING,
  `value_price` DECIMAL(10, 2),
  `value_userId` STRING,
  `value_userSession` STRING,
  -- Define the primary key.
  PRIMARY KEY (`key_id`) NOT ENFORCED,
  ts AS TO_TIMESTAMP_LTZ(value_eventTime, 3),
  WATERMARK FOR ts AS ts - INTERVAL '2' SECOND  -- Define a watermark.
) WITH (
  'connector'='upsert-kafka',
  'topic' = 'click_stream2',
  'properties.bootstrap.servers' = 'alikafka-pre-cn-w******02-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-w******02-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-w******02-3-vpc.alikafka.aliyuncs.com:9092',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'value_',
  'value.fields-include' = 'EXCEPT_KEY'
);



CREATE TEMPORARY TABLE `clickstream3` (
  `key_id` BIGINT,
  `value_eventTime` BIGINT,  
  `value_eventType` STRING,
  `value_productId` STRING,
  `value_categoryId` STRING,
  `value_categoryCode` STRING,
  `value_brand` STRING,
  `value_price` DECIMAL(10, 2),
  `value_userId` STRING,
  `value_userSession` STRING,
  -- Define the primary key.
  PRIMARY KEY (`key_id`) NOT ENFORCED,
  ts AS TO_TIMESTAMP_LTZ(value_eventTime, 3),
  WATERMARK FOR ts AS ts - INTERVAL '2' SECOND  -- Define a watermark.
) WITH (
  'connector'='upsert-kafka',
  'topic' = 'click_stream3',
  'properties.bootstrap.servers' = 'alikafka-pre-cn-w******02-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-w******02-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-w******02-3-vpc.alikafka.aliyuncs.com:9092',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'value_',
  'value.fields-include' = 'EXCEPT_KEY'
);



CREATE TEMPORARY TABLE `clickstream4` (
  `key_id` BIGINT,
  `value_eventTime` BIGINT,  
  `value_eventType` STRING,
  `value_productId` STRING,
  `value_categoryId` STRING,
  `value_categoryCode` STRING,
  `value_brand` STRING,
  `value_price` DECIMAL(10, 2),
  `value_userId` STRING,
  `value_userSession` STRING,
  -- Define the primary key.
  PRIMARY KEY (`key_id`) NOT ENFORCED,
  ts AS TO_TIMESTAMP_LTZ(value_eventTime, 3),
  WATERMARK FOR ts AS ts - INTERVAL '2' SECOND  -- Define a watermark.
) WITH (
  'connector'='upsert-kafka',
  'topic' = 'click_stream4',
  'properties.bootstrap.servers' = 'alikafka-pre-cn-w******02-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-w******02-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-w******02-3-vpc.alikafka.aliyuncs.com:9092',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'value_',
  'value.fields-include' = 'EXCEPT_KEY'
);


CREATE TEMPORARY TABLE `clickstream5` (
  `key_id` BIGINT,
  `value_eventTime` BIGINT,  
  `value_eventType` STRING,
  `value_productId` STRING,
  `value_categoryId` STRING,
  `value_categoryCode` STRING,
  `value_brand` STRING,
  `value_price` DECIMAL(10, 2),
  `value_userId` STRING,
  `value_userSession` STRING,
  -- Define the primary key.
  PRIMARY KEY (`key_id`) NOT ENFORCED,
  ts AS TO_TIMESTAMP_LTZ(value_eventTime, 3),
  WATERMARK FOR ts AS ts - INTERVAL '2' SECOND  -- Define a watermark.
) WITH (
  'connector'='upsert-kafka',
  'topic' = 'click_stream5',
  'properties.bootstrap.servers' = 'alikafka-pre-cn-w******02-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-w******02-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-w******02-3-vpc.alikafka.aliyuncs.com:9092',
  'key.format' = 'json',
  'value.format' = 'json',
  'key.fields-prefix' = 'key_',
  'value.fields-prefix' = 'value_',
  'value.fields-include' = 'EXCEPT_KEY'
);

BEGIN STATEMENT SET; 
INSERT INTO `clickstream1`
SELECT
  id,
  UNIX_TIMESTAMP(eventTime) * 1000 as eventTime,
  eventType,
  productId,
  categoryId,
  categoryCode,
  brand,
  price,
  `userId`,
  userSession
FROM `mysql-catalog`.`ecommerce`.`click_stream1`;


INSERT INTO `clickstream2`
SELECT
  id,
  UNIX_TIMESTAMP(eventTime) * 1000 as eventTime,
  eventType,
  productId,
  categoryId,
  categoryCode,
  brand,
  price,
  `userId`,
  userSession
FROM `mysql-catalog`.`ecommerce`.`click_stream2`;


INSERT INTO `clickstream3`
SELECT
  id,
  UNIX_TIMESTAMP(eventTime) * 1000 as eventTime,
  eventType,
  productId,
  categoryId,
  categoryCode,
  brand,
  price,
  `userId`,
  userSession
FROM `mysql-catalog`.`ecommerce`.`click_stream3`;


INSERT INTO `clickstream4`
SELECT
  id,
  UNIX_TIMESTAMP(eventTime) * 1000 as eventTime,
  eventType,
  productId,
  categoryId,
  categoryCode,
  brand,
  price,
  `userId`,
  userSession
FROM `mysql-catalog`.`ecommerce`.`click_stream4`;


INSERT INTO `clickstream5`
SELECT
  id,
  UNIX_TIMESTAMP(eventTime) * 1000 as eventTime,
  eventType,
  productId,
  categoryId,
  categoryCode,
  brand,
  price,
  `userId`,
  userSession
FROM `mysql-catalog`.`ecommerce`.`click_stream5`;
END;      -- Required when you write data to multiple sinks.

    4. In the upper-right corner, click Deploy to deploy the job.

    5. In the left-side navigation pane, click Operation Center > Job O&M. In the Actions column for the target job, click Start. Then, select Stateless Start and click Start.

    Step 3: Develop, deploy, and start the CEP job

    This topic describes how to deploy the cep-demo-1.0-SNAPSHOT-jar-with-dependencies.jar job, which consumes user clickstream events from Kafka, processes them, and prints alert information to the Realtime Compute for Apache Flink development console. You can adjust the code to suit your business architecture and select an appropriate downstream connector for different data output scenarios. For more information about supported connectors, see Supported connectors.

    1. Code development

    This section provides only the core code and a description of its functionality.

    Main class

    public class CepDemo {

    public static void checkArg(String argName, MultipleParameterTool params) {
        if (!params.has(argName)) {
            throw new IllegalArgumentException(argName + " must be set!");
        }
    }

  // Parses the rule table.
    private static Match_results parseOutput(String output) {
        String rule = "\\(id, version\\): \\((\\d+), (\\d+)\\).*?Event\\((\\d+), (\\w+), (\\d+), (\\d+)";
        Pattern pattern = Pattern.compile(rule);
        Matcher matcher = pattern.matcher(output);
        if (matcher.find()) {
            return new Match_results(Integer.parseInt(matcher.group(1)), Integer.parseInt(matcher.group(2)), Integer.parseInt(matcher.group(3)), matcher.group(4), Integer.parseInt(matcher.group(6)));
        }
        return null;
    }

    public static void main(String[] args) throws Exception {
        // Process args.
        final MultipleParameterTool params = MultipleParameterTool.fromArgs(args);

        // The endpoints of Kafka brokers.
        checkArg(KAFKA_BROKERS_ARG, params);
        // The input topic.
        checkArg(INPUT_TOPIC_ARG, params);
        // The consumer group.
        checkArg(INPUT_TOPIC_GROUP_ARG, params);

        // The JDBC URL of the MySQL database.
        checkArg(JDBC_URL_ARG, params);
        // The name of the MySQL table.
        checkArg(TABLE_NAME_ARG, params);
        // The polling interval for the database.
        checkArg(JDBC_INTERVAL_MILLIS_ARG, params);
        // Specifies whether to use event time for processing (true/false).
        checkArg(USING_EVENT_TIME, params);

        // Set up the streaming execution environment.
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Build Kafka source with new Source API based on FLIP-27. Kafka source settings.
        KafkaSource<Event> kafkaSource =
                KafkaSource.<Event>builder()
                        .setBootstrapServers(params.get(KAFKA_BROKERS_ARG))
                        .setTopics(params.get(INPUT_TOPIC_ARG))
                        .setStartingOffsets(OffsetsInitializer.latest())
                        .setGroupId(params.get(INPUT_TOPIC_GROUP_ARG))
                        .setDeserializer(new EventDeSerializationSchema())
                        .build();



        // DataStream Source. Watermark strategy.
        DataStreamSource<Event> source =
                env.fromSource(
                        kafkaSource,
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((event, ts) -> event.getEventTime()),
                        "Kafka Source");


        // Key the stream by userId for subsequent pattern matching.
        KeyedStream<Event, String> keyedStream =
                source.assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forGenerator(ctx -> new EventBoundedOutOfOrdernessWatermarks(Duration.ofSeconds(5)))
                ).keyBy(new KeySelector<Event, String>() {
                    @Override
                    public String getKey(Event value) throws Exception {
                        return value.getUserId(); // Use only userId as the key.
                    }
                });

        // Dynamic CEP patterns. Use the dynamically loaded pattern processor factory class JDBCPeriodicPatternProcessorDiscovererFactory to retrieve patterns and perform pattern matching (read the rule table in the MySQL database).
        SingleOutputStreamOperator<String> output =
                CEP.dynamicPatterns(
                        keyedStream,// The source data stream.
                        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>() {})
                );

        // Print output to the client.
        output.print();
  
        env.execute("CEP Demo");
    }
}

    Scenario 1: Detect when a user browses products from different categories in the same session within 5 minutes

    Event sequence: The pattern starts with an event of type view, followed by another event of type view.

    public class FirstViewCondition extends SimpleCondition<ClickEvent> {
    @Override
    public boolean filter(ClickEvent event) {
        return event.getEventType().equals("view");
    }
}

    Scenario 2: Detect when a user adds a high-value product to the cart and completes the purchase within 30 minutes

    Event sequence: The pattern starts with an event of type cart (add to cart) with a price greater than 200, followed by an event of type purchase.

    public class CartAddCondition extends SimpleCondition<ClickEvent> {

    @Override
    public boolean filter(ClickEvent event) {
        return event.getEventType().equals("cart") && event.getPrice() > 200;
    }
}
    public class PurchaseCondition extends SimpleCondition<ClickEvent> {

    @Override
    public boolean filter(ClickEvent event) {
        return event.getEventType().equals("purchase");
    }
}

    Scenario 3: Detect when a user browses the same product multiple times within 15 minutes

    Event sequence: The pattern starts with an event of type view, followed by three more events of the same type.

    -- The condition class is the same as in Scenario 1.
public class FirstViewCondition extends SimpleCondition<ClickEvent> {
    @Override
    public boolean filter(ClickEvent event) {
        return event.getEventType().equals("view");
    }
}

    Scenario 4: Detect when a user adds a product to the cart only after its price has dropped

    Event sequence: The pattern starts with an event of type view, followed by another view event for the same product at a lower price, and finally an event of type cart.

    -- The condition class is the same as in Scenario 1.
public class FirstViewCondition extends SimpleCondition<ClickEvent> {
    @Override
    public boolean filter(ClickEvent event) {
        return event.getEventType().equals("view");
    }
}
    public class InitialCondition extends IterativeCondition<ClickEvent> {
    @Override
    public boolean filter(ClickEvent event, Context<ClickEvent> ctx) throws Exception {
        ClickEvent initialView = ctx.getEventsForPattern("initial_view").iterator().next();
        return event.getEventType().equals("view") && event.getProductId().equals(initialView.getProductId()) && event.getPrice() < initialView.getPrice();
    }

}
    public class CartCondition extends SimpleCondition<ClickEvent> {
    @Override
    public boolean filter(ClickEvent event) {
        return event.getEventType().equals("cart");
    }
}

    Scenario 5: Detect when a user browses a product multiple times within a week without placing an order

    Event sequence: The pattern starts with a view event, and no purchase event occurs within the following time window.

    -- The condition class is the same as in Scenario 1.
public class FirstViewCondition extends SimpleCondition<ClickEvent> {
    @Override
    public boolean filter(ClickEvent event) {
        return event.getEventType().equals("view");
    }
}
    -- The condition class is the same as in Scenario 2.
public class PurchaseCondition extends SimpleCondition<ClickEvent> {

    @Override
    public boolean filter(ClickEvent event) {
        return event.getEventType().equals("purchase");
    }
}

    2. Job deployment

    On the Operation Center > Job O&M page, click Deploy Job > JAR Job to deploy five separate stream jobs.

    image

    Parameter description:

    Parameter

    Description

    Example

    Deployment mode

    Stream processing

    Streaming mode

    Deployment name

    The name of the JAR job.

    • Job name for Scenario 1: EcommerceCEPRunner1

    • Job name for Scenario 2: EcommerceCEPRunner2

    • Job name for Scenario 3: EcommerceCEPRunner3

    • Job name for Scenario 4: EcommerceCEPRunner4

    • Job name for Scenario 5: EcommerceCEPRunner5

    Engine version

    The Flink engine version used by the current job.

    The SDK for the code in this topic uses JDK 11. You must select a version that contains jdk11. We recommend that you use the latest VVR version.

    vvr-8.0.11-jdk11-flink-1.17

    JAR URI

    Click the upload icon 上传 on the right to upload the cep-demo-1.0-SNAPSHOT-jar-with-dependencies.jar file.

    oss://xxx/artifacts/namespaces/xxx/cep-demo-1.0-SNAPSHOT-jar-with-dependencies.jar

    Entry point class

    The entry point class of the program.

    com.alibaba.ververica.cep.demo.CepDemo

    Entry point main arguments

    You can pass parameters here and call them in the main method.

    For this example, you need to configure the following parameters:

    • bootstrap.servers: The endpoints of the Kafka cluster.

    • clickstream_topic: The Kafka topic for the clickstream to be consumed.

    • group: The consumer group ID.

    • jdbcUrl: The endpoint of the MySQL service.

    • database: The name of the database.

    • user: The username.

    • password: The user password.

    • tableName: The name of the rule table in MySQL.

    • Configuration for Scenario 1: --kafkaBrokers alikafka-pre-cn-******02-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-******02-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-******02-3-vpc.alikafka.aliyuncs.com:9092 --inputTopic click_stream1 --inputTopicGroup clickstream.consumer --jdbcUrl jdbc:mysql://rm-2********7xr86.mysql.rds.aliyuncs.com:3306/ecommerce?user=flink_rds&password=flink123NN@1 --tableName rds_demo1 --jdbcIntervalMs 3000 --usingEventTime false

    • Configuration for Scenario 2: --kafkaBrokers alikafka-pre-cn-******02-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-******02-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-******02-3-vpc.alikafka.aliyuncs.com:9092 --inputTopic click_stream2 --inputTopicGroup clickstream.consumer --jdbcUrl jdbc:mysql://rm-2********7xr86.mysql.rds.aliyuncs.com:3306/ecommerce?user=flink_rds&password=flink123NN@1 --tableName rds_demo2 --jdbcIntervalMs 3000 --usingEventTime false

    • Configuration for Scenario 3: --kafkaBrokers alikafka-pre-cn-******02-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-******02-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-******02-3-vpc.alikafka.aliyuncs.com:9092 --inputTopic click_stream3 --inputTopicGroup clickstream.consumer --jdbcUrl jdbc:mysql://rm-2********7xr86.mysql.rds.aliyuncs.com:3306/ecommerce?user=flink_rds&password=flink123NN@1 --tableName rds_demo3 --jdbcIntervalMs 3000 --usingEventTime false

    • Configuration for Scenario 4: --kafkaBrokers alikafka-pre-cn-******02-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-******02-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-******02-3-vpc.alikafka.aliyuncs.com:9092 --inputTopic click_stream4 --inputTopicGroup clickstream.consumer --jdbcUrl jdbc:mysql://rm-2********7xr86.mysql.rds.aliyuncs.com:3306/ecommerce?user=flink_rds&password=flink123NN@1 --tableName rds_demo4 --jdbcIntervalMs 3000 --usingEventTime false

    • Configuration for Scenario 5: --kafkaBrokers alikafka-pre-cn-******02-1-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-******02-2-vpc.alikafka.aliyuncs.com:9092,alikafka-pre-cn-******02-3-vpc.alikafka.aliyuncs.com:9092 --inputTopic click_stream5 --inputTopicGroup clickstream.consumer --jdbcUrl jdbc:mysql://rm-2********7xr86.mysql.rds.aliyuncs.com:3306/ecommerce?user=flink_rds&password=flink123NN@1 --tableName rds_demo5 --jdbcIntervalMs 3000 --usingEventTime false

    For more information about deployment, see Deploy a JAR job.

    3. Job startup

    On the Job O&M page, click Start in the Actions column for the target job. Select Stateless Start and then click Start. Sequentially start the jobs for all five scenarios: EcommerceCEPRunner1, EcommerceCEPRunner2, EcommerceCEPRunner3, EcommerceCEPRunner4, and EcommerceCEPRunner5.

    For more information about start configurations, see Start a job.

    Step 4: Query alerts

    Cross-selling and up-selling

    1. Insert the rule data into MySQL.

      -- Test data for the cross-selling and up-selling opportunity scenario
INSERT INTO rds_demo1 (
 `id`,
 `version`,
 `pattern`,
 `function`
) values(
  '1',
   1,
  '{"name":"second_view","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":null,"nodes":[{"name":"second_view","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":{"className":"com.alibaba.ververica.cep.demo.condition.FirstViewCondition","type":"CLASS"},"type":"ATOMIC"},{"name":"first_view","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":{"className":"com.alibaba.ververica.cep.demo.condition.FirstViewCondition","type":"CLASS"},"type":"ATOMIC"}],"edges":[{"source":"first_view","target":"second_view","type":"STRICT"}],"window":{"type":"FIRST_AND_LAST","time":{"unit":"MINUTES","size":5}},"afterMatchStrategy":{"type":"NO_SKIP","patternName":null},"type":"COMPOSITE","version":1}
',
  'com.alibaba.ververica.cep.demo.dynamic.DemoPatternProcessFunction')
;

    2. Insert the test user behavior data into MySQL.

      -- Test data for the cross-selling and up-selling opportunity scenario
INSERT INTO `click_stream1` 
(eventTime, eventType, productId, categoryId, categoryCode, brand, price, userId, userSession) 
VALUES
('2020-01-01 00:01:00.0', 'view', 1005073, 2232732093077520756, 'construction.tools.light', 'samsung', 1130.02, 519698804, '69b5d72f-fd6e-4fed-aa23-1286b2ca89a0'),
('2020-01-01 00:01:03.0', 'view', 1005205, 2232732093077520756, 'apparel.shoes', 'xiaomi', 29.95, 519698804, '69b5d72f-fd6e-4fed-aa23-1286b2ca89a0'),
('2020-01-01 00:01:07.0', 'view', 1005205, 2232732093077520756, 'apparel.shoes.step_ins', 'intel', 167.20, 519698804, '69b5d72f-fd6e-4fed-aa23-1286b2ca89a0'),
('2020-01-01 00:01:08.0', 'view', 1005205, 2232732093077520756, 'appliances.personal.massager', 'samsung', 576.33, 519698804, '69b5d72f-fd6e-4fed-aa23-1286b2ca89a0');

    3. View the results in the logs of the Realtime Compute for Apache Flink development console.

      • In the JobManager log, search for the keyword JDBCPeriodicPatternProcessorDiscoverer to view the latest rule.

        image

      • In the TaskManager, view the Stdout log file. Search for the keyword A match for Pattern of (id, version): (1, 1) to see the matches printed in the log.

        image

    High-value product conversion

    1. Insert the rule data into MySQL.

      -- Test data for the high-value product purchase conversion scenario
INSERT INTO rds_demo2 (
 `id`,
 `version`,
 `pattern`,
 `function`
) values(
  '1',
   1,
  '{"name":"purchase","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":null,"nodes":[{"name":"purchase","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":{"className":"com.alibaba.ververica.cep.demo.condition.PurchaseCondition","type":"CLASS"},"type":"ATOMIC"},{"name":"cart_add","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":{"className":"com.alibaba.ververica.cep.demo.condition.CartAddCondition","type":"CLASS"},"type":"ATOMIC"}],"edges":[{"source":"cart_add","target":"purchase","type":"SKIP_TILL_NEXT"}],"window":{"type":"FIRST_AND_LAST","time":{"unit":"MINUTES","size":30}},"afterMatchStrategy":{"type":"NO_SKIP","patternName":null},"type":"COMPOSITE","version":1}
',
  'com.alibaba.ververica.cep.demo.dynamic.DemoPatternProcessFunction')
;

    2. Insert the test user behavior data into MySQL.

      -- Test data for the high-value product purchase conversion scenario
INSERT INTO `click_stream2` 
(eventTime, eventType, productId, categoryId, categoryCode, brand, price, userId, userSession) 
VALUES
('2020-01-01 01:01:01.0','cart',1002923,2053013555631882655,'electronics.smartphone','huawei',249.30,517014550,'b666b914-8abf-4ebe-b674-aa31a1d0f7ce'),
('2020-01-01 01:11:02.0','cart',1004227,2232732093077520756,'construction.tools.light','apple',892.94,590404850,'057050ba-adca-4c7f-99f3-e87d8f9bbded'),
('2020-01-01 01:11:03.0','purchase',1004227,2232732093077520756,'construction.tools.light','apple',892.94,590404850,'057050ba-adca-4c7f-99f3-e87d8f9bbded');

    3. View the results in the logs of the Realtime Compute for Apache Flink development console.

      • In the JobManager log, search for the keyword JDBCPeriodicPatternProcessorDiscoverer to view the latest rule.

        image

      • In the TaskManager, view the Stdout log file. Search for the keyword A match for Pattern of (id, version): (1, 1) to see the matches printed in the log.

        image

    High-intent user identification

    1. Insert the rule data into MySQL.

      -- Test data for the high-intent user identification scenario
INSERT INTO rds_demo3 (
 `id`,
 `version`,
 `pattern`,
 `function`
) values(
  '1',
   1,
  '{"name":"repeat_view","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["TIMES"],"times":{"from":3,"to":3,"windowTime":null},"untilCondition":null},"condition":null,"nodes":[{"name":"repeat_view","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["TIMES"],"times":{"from":3,"to":3,"windowTime":null},"untilCondition":null},"condition":{"className":"com.alibaba.ververica.cep.demo.condition.FirstViewCondition","type":"CLASS"},"type":"ATOMIC"},{"name":"initial_view","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":{"className":"com.alibaba.ververica.cep.demo.condition.FirstViewCondition","type":"CLASS"},"type":"ATOMIC"}],"edges":[{"source":"initial_view","target":"repeat_view","type":"SKIP_TILL_NEXT"}],"window":{"type":"FIRST_AND_LAST","time":{"unit":"MINUTES","size":15}},"afterMatchStrategy":{"type":"NO_SKIP","patternName":null},"type":"COMPOSITE","version":1}
',
  'com.alibaba.ververica.cep.demo.dynamic.DemoPatternProcessFunction')
;

    2. Insert the test user behavior data into MySQL.

      -- Test data for the high-intent user identification scenario
INSERT INTO `click_stream3` 
(eventTime, eventType, productId, categoryId, categoryCode, brand, price, userId, userSession) 
VALUES
('2020-01-01 02:01:01.0','view',21406322,2232732082063278200,'electronics.clocks','casio',81.03,578858757,'bdf051a8-1594-4630-b93d-2ba62b92d039'),
('2020-01-01 02:01:02.0','view',21406322,2232732082063278200,'electronics.clocks','casio',81.03,578858757,'bdf051a8-1594-4630-b93d-2ba62b92d039'),
('2020-01-01 02:01:03.0','view',21406322,2232732082063278200,'electronics.clocks','casio',81.03,578858757,'bdf051a8-1594-4630-b93d-2ba62b92d039'),
('2020-01-01 02:01:04.0','view',21406322,2232732082063278200,'electronics.clocks','casio',81.03,578858757,'bdf051a8-1594-4630-b93d-2ba62b92d039');

    3. View the results in the logs of the Realtime Compute for Apache Flink development console.

      • In the JobManager log, search for the keyword JDBCPeriodicPatternProcessorDiscoverer to view the latest rule.

        image

      • In the TaskManager, view the Stdout log file. Search for the keyword A match for Pattern of (id, version): (1, 1) to see the matches printed in the log.

        image

    Price-sensitive user engagement

    1. Insert the rule data into MySQL.

      -- Test data for the price-sensitive user engagement scenario
INSERT INTO rds_demo4 (
 `id`,
 `version`,
 `pattern`,
 `function`
) values(
  '1',
   1,
  '{"name":"cart_after_price_drop","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":null,"nodes":[{"name":"cart_after_price_drop","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":{"className":"com.alibaba.ververica.cep.demo.condition.CartCondition","type":"CLASS"},"type":"ATOMIC"},{"name":"view_price_drop","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":{"className":"com.alibaba.ververica.cep.demo.condition.InitialCondition","type":"CLASS"},"type":"ATOMIC"},{"name":"initial_view","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":{"className":"com.alibaba.ververica.cep.demo.condition.FirstViewCondition","type":"CLASS"},"type":"ATOMIC"}],"edges":[{"source":"view_price_drop","target":"cart_after_price_drop","type":"STRICT"},{"source":"initial_view","target":"view_price_drop","type":"STRICT"}],"window":{"type":"FIRST_AND_LAST","time":{"unit":"MINUTES","size":10}},"afterMatchStrategy":{"type":"NO_SKIP","patternName":null},"type":"COMPOSITE","version":1}
',
  'com.alibaba.ververica.cep.demo.dynamic.DemoPatternProcessFunction')
;

    2. Insert the test user behavior data into MySQL.

      -- Test data for the price-sensitive user engagement scenario
INSERT INTO `click_stream4` 
(eventTime, eventType, productId, categoryId, categoryCode, brand, price, userId, userSession) 
VALUES
('2020-01-01 03:01:01.0','view',15200496,2053013553484398879,'appliances.kitchen.toster','uragan',38.87,516616354,'f9233034-08e7-46fb-a1ae-175de0d0de7a'),
('2020-01-01 03:01:02.0','view',15200496,2053013553484398879,'appliances.kitchen.toster','uragan',30.87,516616354,'f9233034-08e7-46fb-a1ae-175de0d0de7a'),
('2020-01-01 03:01:03.0','cart',15200496,2053013553484398879,'appliances.kitchen.toster','uragan',30.87,516616354,'f9233034-08e7-46fb-a1ae-175de0d0de7a');

    3. View the results in the logs of the Realtime Compute for Apache Flink development console.

      • In the JobManager log, search for the keyword JDBCPeriodicPatternProcessorDiscoverer to view the latest rule.

        image

      • In the TaskManager, view the Stdout log file. Search for the keyword A match for Pattern of (id, version): (1, 1) to see the matches printed in the log.

        image

    Churn risk alert

    1. Insert the rule data into MySQL.

      -- Test data for the churn risk scenario
INSERT INTO rds_demo5 (
 `id`,
 `version`,
 `pattern`,
 `function`
) values(
  '1',
   1,
  '{"name":"purchase","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":null,"nodes":[{"name":"purchase","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["SINGLE"],"times":null,"untilCondition":null},"condition":{"className":"com.alibaba.ververica.cep.demo.condition.PurchaseCondition","type":"CLASS"},"type":"ATOMIC"},{"name":"first_view","quantifier":{"consumingStrategy":"SKIP_TILL_NEXT","properties":["LOOPING"],"times":{"from":10,"to":10,"windowTime":null},"untilCondition":null},"condition":{"className":"com.alibaba.ververica.cep.demo.condition.FirstViewCondition","type":"CLASS"},"type":"ATOMIC"}],"edges":[{"source":"first_view","target":"purchase","type":"NOT_NEXT"}],"window":{"type":"FIRST_AND_LAST","time":{"unit":"DAYS","size":7}},"afterMatchStrategy":{"type":"NO_SKIP","patternName":null},"type":"COMPOSITE","version":1}
',
  'com.alibaba.ververica.cep.demo.dynamic.DemoPatternProcessFunction')
;

    2. Insert the test user behavior data into MySQL.

      -- Test data for the churn risk scenario
INSERT INTO `click_stream5` 
(eventTime, eventType, productId, categoryId, categoryCode, brand, price, userId, userSession) 
VALUES
('2020-12-10 00:01:01.0', 'view', 28722181, 2053013555321504139, 'appliances.kitchen.meat_grinder', 'respect', 51.22, 563209914, '657edcb2-767d-458c-9598-714c1e474e09'),
('2020-12-10 00:02:02.0', 'view', 28722181, 2053013555321504139, 'appliances.kitchen.meat_grinder', 'respect', 51.22, 563209914, '657edcb2-767d-458c-9598-714c1e474e09'),
('2020-12-10 00:03:03.0', 'view', 28722181, 2053013555321504139, 'appliances.kitchen.meat_grinder', 'respect', 51.22, 563209914, '657edcb2-767d-458c-9598-714c1e474e09'),
('2020-12-10 00:04:03.0', 'view', 28722181, 2053013555321504139, 'appliances.kitchen.meat_grinder', 'respect', 51.22, 563209914, '657edcb2-767d-458c-9598-714c1e474e09'),
('2020-12-10 00:05:04.0', 'view', 28722181, 2053013555321504139, 'appliances.kitchen.meat_grinder', 'respect', 51.22, 563209914, '657edcb2-767d-458c-9598-714c1e474e09'),
('2020-12-10 00:06:05.0', 'view', 28722181, 2053013555321504139, 'appliances.kitchen.meat_grinder', 'respect', 51.22, 563209914, '657edcb2-767d-458c-9598-714c1e474e09'),
('2020-12-10 00:07:06.0', 'view', 28722181, 2053013555321504139, 'appliances.kitchen.meat_grinder', 'respect', 51.22, 563209914, '657edcb2-767d-458c-9598-714c1e474e09'),
('2020-12-10 00:08:07.0', 'view', 28722181, 2053013555321504139, 'appliances.kitchen.meat_grinder', 'respect', 51.22, 563209914, '657edcb2-767d-458c-9598-714c1e474e09'),
('2020-12-10 00:08:08.0', 'view', 28722181, 2053013555321504139, 'appliances.kitchen.meat_grinder', 'respect', 51.22, 563209914, '657edcb2-767d-458c-9598-714c1e474e09'),
('2020-12-10 00:08:08.0', 'view', 28722181, 2053013555321504139, 'appliances.kitchen.meat_grinder', 'respect', 51.22, 563209914, '657edcb2-767d-458c-9598-714c1e474e09'),
('2020-12-10 00:09:09.0', 'cart', 28722181, 2053013555321504139, 'appliances.kitchen.meat_grinder', 'respect', 51.22, 563209914, '657edcb2-767d-458c-9598-714c1e474e09');

    3. View the results in the logs of the Realtime Compute for Apache Flink development console.

      • In the JobManager log, search for the keyword JDBCPeriodicPatternProcessorDiscoverer to view the latest rule.

        image

      • In the TaskManager, view the Stdout log file. Search for the keyword A match for Pattern of (id, version): (1, 1) to see the matches printed in the log.

        image

    Related documents