Flink 电商实战案例：实时类目GMV统计

m0_37670860

443人浏览 · 2026-06-23 09:07:27

m0_37670860 · 2026-06-23 09:07:27 发布

一、项目简介

本文手把手实现一套生产级可落地的 Flink 电商实时计算案例，基于 Flink 1.18 版本、Java 语言开发。

核心业务：实时采集电商订单流，关联商品维表，按类目5秒滚动窗口统计实时GMV、下单量、支付订单数。

本案例解决新手常见痛点：

杜绝错误的「系统时间代替窗口时间」写法

实战 广播流维表JOIN（电商最常用宽表拼接方案）

事件时间+水印处理乱序数据

增量聚合+窗口函数分层计算（高性能）

支持控制台打印 + 实时写入MySQL数据库双模式

开启Checkpoint，支持Exactly-Once语义

二、技术栈

Flink 1.18.0（批流一体）

Java 8

事件时间 + 水印（乱序数据处理）

广播流 Broadcast（维表实时关联）

TumblingEventTimeWindow 滚动窗口

Aggregate + WindowFunction 组合增量聚合

Flink JdbcSink（生产级数据库写入）

Checkpoint 容错恢复、 Exactly-Once 入库

三、业务流程架构

模拟电商实时数据流链路：

自定义数据源持续生成模拟用户下单订单流
数据清洗：过滤无效订单、用户信息脱敏
商品维表通过广播流全局广播
订单流关联广播维表，生成订单商品宽表
按商品类目分组，5秒事件时间滚动窗口聚合
统计各类目：总下单数、支付订单数、区间GMV
结果支持控制台格式化打印 + 实时写入MySQL数据库

四、完整工程代码（新增数据库写入能力）

1. Maven 依赖 pom.xml（新增MySQL驱动、JDBC依赖）

xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.ecommerce</groupId>
    <artifactId>flink-order-real</artifactId>
    <version>1.0.0</version>

    <properties>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <flink.version>1.18.0</flink.version>
        <scala.binary.version>2.12</scala.binary.version>
        <mysql.version>8.0.33</mysql.version>
    </properties>

    <dependencies>
        
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-runtime-web</artifactId>
            <version>${flink.version}</version>
        </dependency>

        
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-jdbc</artifactId>
            <version>${flink.version}</version>
        </dependency>

        
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>${mysql.version}</version>
            <scope>runtime</scope>
        </dependency>

        
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.24</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.8.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <encoding>UTF-8</encoding>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

2. MySQL 建表语句（提前执行）

新建数据库 flink_db，创建类目GMV统计表，用于接收Flink实时写入数据：

sql
CREATE DATABASE IF NOT EXISTS flink_db DEFAULT CHARACTER SET utf8mb4;
USE flink_db;

CREATE TABLE IF NOT EXISTS category_gmv_stat (
    id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT '主键ID',
    category VARCHAR(50) NOT NULL COMMENT '商品类目',
    window_start BIGINT NOT NULL COMMENT '窗口开始时间戳',
    window_end BIGINT NOT NULL COMMENT '窗口结束时间戳',
    total_order_count INT NOT NULL DEFAULT 0 COMMENT '总下单数',
    pay_order_count INT NOT NULL DEFAULT 0 COMMENT '支付订单数',
    total_gmv DECIMAL(10,2) NOT NULL DEFAULT 0.00 COMMENT '区间GMV',
    create_time DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '数据写入时间',
    UNIQUE KEY uk_window_category (window_start, window_end, category)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT 'Flink实时类目GMV统计表';

唯一索引作用：配合Flink Checkpoint，避免窗口数据重复入库，保证幂等性。

3. 实体类层（无改动，直接复用）

Order.java 订单实体

java
package com.ecommerce.entity;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.sql.Timestamp;

@Data
@NoArgsConstructor
@AllArgsConstructor
public class Order {
    private String orderId;
    private String userId;
    private Long goodsId;
    private Double amount;
    private Integer status; // 0未支付 1已支付 2取消 3退款
    private Timestamp orderTime;
}

GoodsInfo.java 商品维表实体

OrderWithGoods.java 订单商品宽表

java
package com.ecommerce.entity;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@NoArgsConstructor
@AllArgsConstructor
public class OrderWithGoods {
    private String orderId;
    private String userId;
    private Long goodsId;
    private String goodsName;
    private String category;
    private Double amount;
    private Integer status;
    private Timestamp orderTime;
}

OrderWindowResult.java 窗口聚合结果实体

java
package com.ecommerce.entity;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

@Data
@NoArgsConstructor
@AllArgsConstructor
public class OrderWindowResult {
    private String category;
    private Long windowStart;
    private Long windowEnd;
    private Long totalOrderCount;
    private Long payOrderCount;
    private Double totalGmv;
}

4. 自定义数据源（无改动，直接复用）

OrderSource.java 实时订单生成器

java
package com.ecommerce.source;

import com.ecommerce.entity.Order;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.sql.Timestamp;
import java.util.Random;

public class OrderSource implements SourceFunction<Order> {
    private volatile boolean isRunning = true;
    private final Random random = new Random();

    @Override
    public void run(SourceContext<Order> ctx) throws Exception {
        while (isRunning) {
            String orderId = "ORD_" + System.currentTimeMillis() + random.nextInt(1000);
            String userId = "USER_" + random.nextInt(1000);
            Long goodsId = (long) random.nextInt(10);
            Double amount = 10.0 + random.nextDouble() * 990;
            Integer status = random.nextInt(4);
            Timestamp now = new Timestamp(System.currentTimeMillis());

            ctx.collect(new Order(orderId, userId, goodsId, amount, status, now));
            Thread.sleep(800);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}

GoodsSource.java 商品维表数据源

java
package com.ecommerce.source;

import com.ecommerce.entity.GoodsInfo;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.HashMap;
import java.util.Map;

public class GoodsSource implements SourceFunction<GoodsInfo> {
    private volatile boolean isRunning = true;
    private static final Map<Long, GoodsInfo> GOODS_MAP = new HashMap<>();

    static {
        GOODS_MAP.put(1L, new GoodsInfo(1L, "iPhone 15", "手机数码"));
        GOODS_MAP.put(2L, new GoodsInfo(2L, "小米14", "手机数码"));
        GOODS_MAP.put(3L, new GoodsInfo(3L, "纯棉T恤", "服饰"));
        GOODS_MAP.put(4L, new GoodsInfo(4L, "运动跑鞋", "服饰"));
        GOODS_MAP.put(5L, new GoodsInfo(5L, "机械键盘", "电脑外设"));
        GOODS_MAP.put(6L, new GoodsInfo(6L, "电竞鼠标", "电脑外设"));
        GOODS_MAP.put(7L, new GoodsInfo(7L, "牛奶一箱", "食品生鲜"));
        GOODS_MAP.put(8L, new GoodsInfo(8L, "零食大礼包", "食品生鲜"));
        GOODS_MAP.put(9L, new GoodsInfo(9L, "护发素", "美妆个护"));
        GOODS_MAP.put(0L, new GoodsInfo(0L, "粉底液", "美妆个护"));
    }

    @Override
    public void run(SourceContext<GoodsInfo> ctx) throws Exception {
        while (isRunning) {
            for (GoodsInfo value : GOODS_MAP.values()) {
                ctx.collect(value);
            }
            Thread.sleep(5000);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}

5. 核心主程序（新增MySQL JdbcSink入库逻辑）

保留原有控制台打印，新增批量、容错、幂等的MySQL入库逻辑，完全适配本地/集群运行：

java
package com.ecommerce;

import com.ecommerce.entity.GoodsInfo;
import com.ecommerce.entity.Order;
import com.ecommerce.entity.OrderWithGoods;
import com.ecommerce.entity.OrderWindowResult;
import com.ecommerce.source.GoodsSource;
import com.ecommerce.source.OrderSource;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.time.Duration;
import java.sql.PreparedStatement;

public class EcommerceOrderJob {
    // MySQL数据库连接参数（根据自己环境修改）
    private static final String MYSQL_URL = "jdbc:mysql://localhost:3306/flink_db?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai&rewriteBatchedStatements=true";
    private static final String MYSQL_USER = "root";
    private static final String MYSQL_PASSWORD = "123456";

    public static void main(String[] args) throws Exception {
        // 1. 初始化流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 开启Checkpoint，保障Exactly-Once一致性（数据库入库必须开启）
        env.enableCheckpointing(3000);

        // 2. 订单流 + 事件时间水印（处理2秒内乱序数据）
        DataStream<Order> orderStream = env.addSource(new OrderSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                                .withTimestampAssigner((order, ts) -> order.getOrderTime().getTime())
                );

        // 3. 数据清洗：过滤无效订单、用户ID脱敏
        SingleOutputStreamOperator<Order> cleanOrderStream = orderStream
                .filter(order -> order.getAmount() > 0)
                .map(new MapFunction<Order, Order>() {
                    @Override
                    public Order map(Order order) {
                        String desensitizeUserId = order.getUserId().substring(0, 5) + "****";
                        order.setUserId(desensitizeUserId);
                        return order;
                    }
                });

        // 4. 商品维表广播配置
        MapStateDescriptor<Long, GoodsInfo> goodsStateDesc = new MapStateDescriptor<>(
                "goods_broadcast_state",
                Types.LONG,
                Types.POJO(GoodsInfo.class)
        );
        DataStream<GoodsInfo> goodsStream = env.addSource(new GoodsSource());
        BroadcastStream<GoodsInfo> broadcastGoodsStream = goodsStream.broadcast(goodsStateDesc);

        // 5. 订单流关联广播维表，生成订单商品宽表
        SingleOutputStreamOperator<OrderWithGoods> orderWideStream = cleanOrderStream.connect(broadcastGoodsStream)
                .process(new BroadcastProcessFunction<Order, GoodsInfo, OrderWithGoods>() {
                    @Override
                    public void processElement(Order order, ReadOnlyContext ctx, Collector<OrderWithGoods> out) throws Exception {
                        GoodsInfo goods = ctx.getBroadcastState(goodsStateDesc).get(order.getGoodsId());
                        if (goods != null) {
                            OrderWithGoods wide = new OrderWithGoods();
                            wide.setOrderId(order.getOrderId());
                            wide.setUserId(order.getUserId());
                            wide.setGoodsId(order.getGoodsId());
                            wide.setGoodsName(goods.getGoodsName());
                            wide.setCategory(goods.getCategory());
                            wide.setAmount(order.getAmount());
                            wide.setStatus(order.getStatus());
                            wide.setOrderTime(order.getOrderTime());
                            out.collect(wide);
                        }
                    }

                    @Override
                    public void processBroadcastElement(GoodsInfo goods, Context ctx, Collector<OrderWithGoods> out) throws Exception {
                        ctx.getBroadcastState(goodsStateDesc).put(goods.getGoodsId(), goods);
                    }
                });

        // 6. 逐条打印实时订单明细
        orderWideStream.map(order -> {
            System.out.println("【单条订单明细】" + order);
            return order;
        });

        // 7. 5秒事件时间滚动窗口，按类目聚合GMV
        SingleOutputStreamOperator<OrderWindowResult> windowAggStream = orderWideStream
                .keyBy(OrderWithGoods::getCategory)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                // 增量聚合 + 窗口函数组合（高性能+获取真实窗口时间）
                .aggregate(new OrderAggFunc(), new OrderWindowFunc());

        // 8. 自定义Sink控制台格式化打印
        windowAggStream.addSink(new CustomConsoleSink());

        // 9. 新增：Flink JdbcSink 实时写入MySQL（生产级入库）
        String insertSql = "INSERT INTO category_gmv_stat (category, window_start, window_end, total_order_count, pay_order_count, total_gmv) VALUES (?,?,?,?,?) " +
                "ON DUPLICATE KEY UPDATE total_order_count=VALUES(total_order_count),pay_order_count=VALUES(pay_order_count),total_gmv=VALUES(total_gmv)";

        // 配置JDBC批量执行参数
        JdbcExecutionOptions executionOptions = JdbcExecutionOptions.builder()
                .withBatchSize(100) // 每100条批量写入
                .withBatchIntervalMs(1000) // 最大1秒刷写一次
                .withMaxRetries(3) // 失败重试3次
                .build();

        // 构建JdbcSink
        SinkFunction<OrderWindowResult> mysqlSink = JdbcSink.sink(
                insertSql,
                (PreparedStatement statement, OrderWindowResult result) -> {
                    statement.setString(1, result.getCategory());
                    statement.setLong(2, result.getWindowStart());
                    statement.setLong(3, result.getWindowEnd());
                    statement.setLong(4, result.getTotalOrderCount());
                    statement.setLong(5, result.getPayOrderCount());
                    statement.setBigDecimal(6, new BigDecimal(result.getTotalGmv()).setScale(2, BigDecimal.ROUND_HALF_UP));
                },
                executionOptions,
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl(MYSQL_URL)
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .withUsername(MYSQL_USER)
                        .withPassword(MYSQL_PASSWORD)
                        .build()
        );

        // 绑定MySQL入库Sink
        windowAggStream.addSink(mysqlSink);

        env.execute("电商实时类目GMV统计任务-MySQL入库版");
    }

    /**
     * 增量聚合函数：轻量化累加计算
     * 累加器：Tuple2<Tuple2<总订单数,总GMV>, 支付订单数>
     */
    public static class OrderAggFunc implements AggregateFunction<OrderWithGoods, Tuple2<Tuple2<Long, Double>, Long>, Tuple2<Tuple2<Long, Double>, Long>> {
        @Override
        public Tuple2<Tuple2<Long, Double>, Long> createAccumulator() {
            return Tuple2.of(Tuple2.of(0L, 0.0D), 0L);
        }

        @Override
        public Tuple2<Tuple2<Long, Double>, Long> add(OrderWithGoods order, Tuple2<Tuple2<Long, Double>, Long> acc) {
            acc.f0.f0 += 1;
            acc.f0.f1 += order.getAmount();
            if (order.getStatus() == 1) {
                acc.f1 += 1;
            }
            return acc;
        }

        @Override
        public Tuple2<Tuple2<Long, Double>, Long> getResult(Tuple2<Tuple2<Long, Double>, Long> acc) {
            return acc;
        }

        @Override
        public Tuple2<Tuple2<Long, Double>, Long> merge(Tuple2<Tuple2<Long, Double>, Long> acc1, Tuple2<Tuple2<Long, Double>, Long> acc2) {
            return Tuple2.of(
                    Tuple2.of(acc1.f0.f0 + acc2.f0.f0, acc1.f0.f1 + acc2.f0.f1),
                    acc1.f1 + acc2.f1
            );
        }
    }

    /**
     * 窗口函数：获取真实事件时间窗口，修复系统时间BUG
     */
    public static class OrderWindowFunc implements WindowFunction<Tuple2<Tuple2<Long, Double>, Long>, OrderWindowResult, String, TimeWindow> {
        @Override
        public void apply(String category,
                          TimeWindow window,
                          Iterable<Tuple2<Tuple2<Long, Double>, Long>> input,
                          Collector<OrderWindowResult> out) throws Exception {
            Tuple2<Tuple2<Long, Double>, Long> acc = input.iterator().next();

            OrderWindowResult result = new OrderWindowResult();
            result.setCategory(category);
            // 核心：使用事件时间窗口原生时间，不依赖机器系统时间
            result.setWindowStart(window.getStart());
            result.setWindowEnd(window.getEnd());
            result.setTotalOrderCount(acc.f0.f0);
            result.setPayOrderCount(acc.f1);
            result.setTotalGmv(acc.f0.f1);

            out.collect(result);
        }
    }

    /**
     * 自定义控制台Sink：格式化输出窗口统计结果
     */
    public static class CustomConsoleSink implements SinkFunction<OrderWindowResult> {
        @Override
        public void invoke(OrderWindowResult result, Context context) throws Exception {
            System.out.println("\n====================【类目5秒GMV汇总】====================");
            System.out.printf("商品类目：%s%n", result.getCategory());
            System.out.printf("窗口时间：%d ~ %d%n", result.getWindowStart(), result.getWindowEnd());
            System.out.printf("总下单订单数：%d 笔%n", result.getTotalOrderCount());
            System.out.printf("成功支付订单：%d 笔%n", result.getPayOrderCount());
            System.out.printf("区间累计GMV：%.2f 元%n", result.getTotalGmv());
            System.out.println("【数据已实时写入MySQL数据库】");
            System.out.println("=========================================================\n");
        }
    }
}

五、核心入库知识点详解

1. 为什么用 JdbcSink 不用原生自定义Sink？

支持批量写入：单条写入性能极差，JdbcSink 支持批量刷库，适配大数据量

支持失败重试：网络抖动、数据库卡顿自动重试，避免数据丢失

支持Exactly-Once：配合Checkpoint + 唯一索引，彻底解决重复数据问题

官方维护：生产环境标准方案，稳定无BUG

2. 幂等性入库核心

代码中使用 ON DUPLICATE KEY UPDATE 语法 + 数据库唯一索引：

窗口触发、Checkpoint重试、任务重启时，不会重复插入数据

重复窗口数据会自动覆盖更新，保证数据最终一致性

3. 本地运行 & 服务器集群运行日志区别

本地IDE运行：System.out.println 打印在IDE控制台，数据直接写入本地MySQL

Flink集群运行：控制台打印日志输出到 taskmanager.out 文件，数据库写入逻辑完全不变，无需改代码

六、运行步骤

本地MySQL创建 flink_db 数据库，执行上方建表语句
修改代码中 MySQL 账号、密码、地址为自己的环境
启动程序，控制台打印统计数据，同时自动入库MySQL
执行SQL查询实时数据：SELECT * FROM category_gmv_stat;

java
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

Flink 机制：

本地 IDE 运行时：自动创建 本地迷你单节点环境（LocalEnvironment）

不连接任何远程 Flink 集群

不依赖服务器、不依赖 Yarn、不依赖 Standalone 集群

所有 Task、JobManager、TaskManager 全部在你本地 JVM 进程内模拟

代码里没有任何集群配置

没有指定集群地址

没有提交 Yarn/Standalone 集群

没有打包 Job

如何改成【真正的 Flink 服务器集群执行】？

只需要 2 步，企业生产标准操作：

1. 代码无需改动（核心！！）

Flink 代码不用改一行
Flink 设计理念：一套代码，本地调试、集群运行无缝切换

2. 部署方式改变

步骤 1：maven 打包成 jar

bash
mvn clean package

步骤 2：服务器集群命令提交

bash
# Standalone 集群提交
flink run -m flink集群IP:8081 -p 1 flink-order-real-1.0.0.jar

# Yarn 集群提交
flink run -yjm 1024m -ytm 2048m flink-order-real-1.0.0.jar

此时任务就会跑在远端 Flink 服务器集群，由集群的 JobManager、TaskManager 调度执行。

七、生产环境优化方案