基于FlinkCEP的电商营销实时预警系统

Pattern<ClickEvent, ?> crossSellPattern = Pattern.<ClickEvent>begin("first_view")
         // 模式开始于一个事件类型为"view"的ClickEvent
        .where(new SimpleCondition<ClickEvent>() {
            @Override
            public boolean filter(ClickEvent event) {
                return event.getEventType().equals("view");
            }
        })
        // 接下来是一个事件类型同样为"view"的ClickEvent
        .next("second_view")
        .where(new SimpleCondition<ClickEvent>() {
            @Override
            public boolean filter(ClickEvent event) {
                return event.getEventType().equals("view");
            }
        })
        // 整个模式必须在5分钟内完成
        .within(Time.minutes(5));

SingleOutputStreamOperator<Alert> crossSellAlerts = crossSellPatternStream.select(new PatternSelectFunction<ClickEvent, Alert>() {
    @Override
    public Alert select(Map<String, List<ClickEvent>> pattern) {
        ClickEvent firstView = pattern.get("first_view").get(0);
        ClickEvent secondView = pattern.get("second_view").get(0);

        // 检查类别是否不同
        if (!firstView.getCategoryCode().equals(secondView.getCategoryCode())) {
            var message = "Cross-sell opportunity detected for user " + firstView.getUserId() +
                    ": viewed products in categories " + firstView.getCategoryCode() + " and " + secondView.getCategoryCode();
            // 返回交叉销售与追加销售机会预警
            return new Alert(secondView.getUserSession(), secondView.getUserId(), AlertType.CROSS_UPSELL, message);
        }
        return null; // 如果类别相同则返回null，有效地将其过滤掉
    }
}).filter(Objects::nonNull).name("CrossSellAlertsPattern").uid("CrossSellAlertsPattern");

Pattern<ClickEvent, ?> abandonmentPattern = Pattern.<ClickEvent>begin("cart_add")
        // 模式开始于一个事件类型为"cart"且价格大于200的ClickEvent
        .where(new SimpleCondition<ClickEvent>() {
            @Override
            public boolean filter(ClickEvent event) {
                return event.getEventType().equals("cart") && event.getPrice() > 200;
            }
        })
        // 接下来是一个事件类型为"purchase"的ClickEvent
        .followedBy("purchase")
        .where(new SimpleCondition<ClickEvent>() {
            @Override
            public boolean filter(ClickEvent event) {
                return event.getEventType().equals("purchase");
            }
        })
        // 整个模式必须在10分钟内完成
        .within(Time.minutes(10));

SingleOutputStreamOperator<Alert> abandonedmentAlert = abandonmentPatternStream
        .select(new PatternTimeoutFunction<ClickEvent, Alert>() {
            @Override
            // 处理超时事件
            public Alert timeout(Map<String, List<ClickEvent>> pattern, long timeoutTimestamp) {
                ClickEvent cartAdd = pattern.get("cart_add").get(0);

                var message = "High-value cart abandonment detected for user '" + cartAdd.getUserId() + "' priced at " + cartAdd.getPrice() +
                        ". No purchase within 10 minutes.";
                // 返回高价值购物车挽回预警
                return new Alert(cartAdd.getUserSession(), cartAdd.getUserId(), AlertType.CART_ABANDONMENT, message);
            }
        }, new PatternSelectFunction<ClickEvent, Alert>() {
            @Override
             // 处理匹配的模式
            public Alert select(Map<String, List<ClickEvent>> pattern) {
                ClickEvent cartAdd = pattern.get("cart_add").get(0);
                ClickEvent purchase = pattern.get("purchase").get(0);

                var message = "Purchase completed for user " + purchase.getUserId() + " on product " + purchase.getProductId() +
                        " priced at " + purchase.getPrice();
                // 返回购买完成预警
                return new Alert(cartAdd.getUserSession(), cartAdd.getUserId(), AlertType.PURCHASE_COMPLETION, message);
            }
        }).map(new MapFunction<Either<Alert, Alert>, Alert>() {
            @Override
            public Alert map(Either<Alert, Alert> alert) throws Exception {
                if (alert.isLeft()) {
                    // 如果是超时预警，返回左值
                    return alert.left();
                } else {
                    // 如果是匹配预警，返回右值
                    return alert.right();
                }
            }
        }).name("AbandonmentAlertsPattern").uid("AbandonmentAlertsPattern");

Pattern<ClickEvent, ?> purchaseIntentPattern = Pattern.<ClickEvent>begin("initial_view")
        // 模式开始于一个事件类型为"view"的ClickEvent
        .where(new SimpleCondition<ClickEvent>() {
            @Override
            public boolean filter(ClickEvent event) {
                return event.getEventType().equals("view");
            }
        })
        // 接下来是一个事件类型同样为"view"的ClickEvent，并且这个事件重复3次
        .followedBy("repeat_view")
        .where(new SimpleCondition<ClickEvent>() {
            @Override
            public boolean filter(ClickEvent event) {
                return event.getEventType().equals("view");
            }
        }).times(3)
        // 整个模式必须在15分钟内完成
        .within(Time.minutes(15));

SingleOutputStreamOperator<Alert> purchaseIntentAlerts = purchaseIntentStream.select(new PatternSelectFunction<ClickEvent, Alert>() {
    @Override
    public Alert select(Map<String, List<ClickEvent>> pattern) {
        ClickEvent initialView = pattern.get("initial_view").get(0);
        var message = "High purchase intent detected for user " + initialView.getUserId() +
                " on product " + initialView.getProductId();
        // 返回高意向用户识别预警
        return new Alert(initialView.getUserSession(), initialView.getUserId(), AlertType.PRICE_SENSITIVITY, message);
    }
}).name("PurchaseIntentAlertsPattern").uid("PurchaseIntentAlertsPattern");

Pattern<ClickEvent, ?> priceSensitivityPattern = Pattern.<ClickEvent>begin("initial_view")
        // 模式开始于一个事件类型为"view"的ClickEvent
        .where(new SimpleCondition<ClickEvent>() {
            @Override
            public boolean filter(ClickEvent event) {
                return event.getEventType().equals("view");
            }
        })
        // 接下来是一个事件类型同样为"view"的ClickEvent，并且该产品的价格低于初始产品的价格
        .next("view_price_drop")
        .where(new 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();
            }
        })
        // 接下来是一个事件类型为"cart"的ClickEvent
        .next("cart_after_price_drop")
        .where(new SimpleCondition<ClickEvent>() {
            @Override
            public boolean filter(ClickEvent event) {
                return event.getEventType().equals("cart");
            }
        })
        // 整个模式必须在10分钟内完成
        .within(Time.minutes(10));

SingleOutputStreamOperator<Alert> priceSensitivityAlerts = priceSensitivityStream.select(new PatternSelectFunction<ClickEvent, Alert>() {
    @Override
    public Alert select(Map<String, List<ClickEvent>> pattern) {
        ClickEvent initialView = pattern.get("initial_view").get(0);
        var message = "Price-sensitive customer detected for user " + initialView.getUserId() +
                " on product " + initialView.getProductId() + " after a price drop.";
        // 返回价格敏感用户运营预警
        return new Alert(initialView.getUserSession(), initialView.getUserId(), AlertType.PRICE_SENSITIVITY, message);
    }
}).name("PriceSensitivityAlertsPattern").uid("PriceSensitivityAlertsPattern");

Pattern<ClickEvent, ?> churnPredictionPattern = Pattern.<ClickEvent>begin("first_view")
        // 模式开始于一个事件类型为"view"的ClickEvent
        .where(new SimpleCondition<ClickEvent>() {
            @Override
            public boolean filter(ClickEvent event) {
                return event.getEventType().equals("view");
            }
        })
         // 用户必须至少查看10次
        .timesOrMore(10)  
        // 在接下来的时间窗口内不能有"purchase"事件
        .notNext("purchase")
        .where(new SimpleCondition<ClickEvent>() {
            @Override
            public boolean filter(ClickEvent event) {
                return event.getEventType().equals("purchase");
            }
        })
        // 整个模式必须在7天内完成
        .within(Time.days(7));

SingleOutputStreamOperator<Alert> churnPredictionAlerts = churnPredictionStream.select(new PatternSelectFunction<ClickEvent, Alert>() {
    @Override
    public Alert select(Map<String, List<ClickEvent>> pattern) {
        ClickEvent firstView = pattern.get("first_view").get(0);
        var message = "Churn risk detected for user " + firstView.getUserId() +
                ": viewed multiple products over the week without making a purchase.";
        // 返回流失风险预警
        return new Alert(firstView.getUserSession(), firstView.getUserId(), AlertType.CHURN_RISK, message);
    }
}).name("ChurnPredictionAlertsPattern").uid("ChurnPredictionAlertsPattern");