基于 Flink 的订单过滤与金额统计

1282 字

6 分钟

基于 Flink 的订单过滤与金额统计

2026-06-06

课程实验

Flink

/

大数据

/

流处理

/

Java

/

Python

项目简介#

这是一个大数据技术基础课程的实验项目，核心目标是用 Apache Flink 实现实时流处理——对订单数据进行过滤和金额聚合统计。

整个系统由三部分组成：

Web 前端 — 输入订单数据，实时展示统计结果
Python 桥接服务器 — 充当 HTTP ↔ TCP 的桥梁，连接前端和 Flink
Flink 流处理引擎 — 执行真正的数据过滤和聚合计算

数据流如下：

1
浏览器 → HTTP POST → Python Server → TCP Socket → Flink
2
                                                    ↓
3
                                            flatMap → filter
4
                                            → keyBy → sum
5
                                                    ↓
6
                                                  输出

处理流程#

业务逻辑很简单：接收格式为 商品名金额 的订单数据，过滤掉金额 < 3 的无效订单，然后按商品名分组累加金额。

Flink 的算子链：

1
DataStream<String> socketStream = env.addSource(new RetrySocketSource("localhost", 7777));
2

3
DataStream<Tuple2<String, Double>> result = socketStream
4
    .flatMap(new OrderParser())        // 拆分字段："苹果 5" → ("苹果", 5.0)
5
    .filter(new ValidOrderFilter())    // 过滤金额 < 3
6
    .keyBy(tuple -> tuple.f0)          // 按商品名分组
7
    .sum(1);                           // 金额求和
8

9
result.print("有效订单统计");

每个算子的职责：

算子	作用
`flatMap`	将 `"苹果 5"` 拆分为 `Tuple2("苹果", 5.0)`
`filter`	过滤掉金额 < 3 的无效订单
`keyBy`	按商品名分组，相同商品路由到同一任务
`sum`	对每组的金额累加求和，实时输出更新结果

以一组测试数据为例：

1
原始数据        flatMap           filter(≥3)        keyBy+sum
2
─────────      ─────────        ──────────        ──────────
3
苹果 5    →  (苹果, 5.0)   →   (苹果, 5.0)   →   (苹果, 5.0)
4
香蕉 2    →  (香蕉, 2.0)   →   [已过滤]      →   -
5
苹果 10   →  (苹果, 10.0)  →   (苹果, 10.0)  →   (苹果, 15.0)
6
橘子 1    →  (橘子, 1.0)   →   [已过滤]      →   -
7
香蕉 8    →  (香蕉, 8.0)   →   (香蕉, 8.0)   →   (香蕉, 8.0)
8
苹果 3    →  (苹果, 3.0)   →   (苹果, 3.0)   →   (苹果, 18.0)
9
橘子 6    →  (橘子, 6.0)   →   (橘子, 6.0)   →   (橘子, 6.0)

踩坑：自定义重连数据源#

Flink 自带的 socketTextStream 有个 maxRetry 参数，理论上可以控制重连次数。但在 Flink 1.17.1 中实测发现这个参数并不靠谱——连接失败后不会按预期重试。

于是自己写了一个 RetrySocketSource，继承 SourceFunction<String>，核心逻辑就是死循环 + 3 秒重试：

1
public class RetrySocketSource extends SourceFunction<String> {
2
    private volatile boolean isRunning = true;
3

4
    @Override
5
    public void run(SourceContext<String> ctx) throws Exception {
6
        while (isRunning) {
7
            try (Socket socket = new Socket(host, port);
8
                 BufferedReader reader = new BufferedReader(
9
                     new InputStreamReader(socket.getInputStream()))) {
10
                String line;
11
                while (isRunning && (line = reader.readLine()) != null) {
12
                    ctx.collect(line);
13
                }
14
            } catch (Exception e) {
15
                if (isRunning) {
16
                    Thread.sleep(3000); // 3秒后重试
17
                }
18
            }
19
        }
20
    }
21
}

这个方案虽然朴素，但非常实用——无论 TCP 服务器什么时候启动，Flink 都能自动连上。

Python 桥接服务器#

为了让浏览器能和 Flink 通信，用 Python 标准库写了一个双协议服务器：

HTTP 服务（端口 8080）：提供前端页面 + 接收 /api/order 的 POST 请求
TCP 服务（端口 7777）：等待 Flink 的 Socket 连接

核心思路：前端提交订单 → HTTP handler 收到后 → 转发给已连接的 Flink TCP 客户端。

1
# 线程安全的客户端管理
2
clients = []
3
clients_lock = threading.Lock()
4

5
def broadcast(message):
6
    with clients_lock:
7
        dead = []
8
        for client in clients:
9
            try:
10
                client.sendall((message + '\n').encode('utf-8'))
11
            except:
12
                dead.append(client)
13
        for c in dead:
14
            clients.remove(c)

没有用 Flask、Django 这些框架，纯粹用 http.server + socket + threading，零依赖，拿来跑实验刚刚好。

Web 前端#

前端是一个单页应用，纯 HTML/CSS/JS，主要功能：

订单输入表单 + 快速填充按钮
实时统计面板（总订单数、有效订单数、已过滤数、有效总金额）
商品汇总表 + 订单历史记录
颜色标签区分有效/已过滤订单

有意思的一点：前端也独立实现了 amount >= 3 的过滤逻辑，这样用户在提交订单的瞬间就能看到颜色标记，不用等 Flink 的输出回来——流处理链路的延迟虽然不大，但本地判定能让交互反馈做到零等待。

技术栈总结#

层	技术
流处理引擎	Apache Flink 1.17.1（DataStream API）
后端桥接	Python 3 标准库（http.server + socket + threading）
前端	原生 HTML/CSS/JS
构建工具	Maven + maven-shade-plugin（打包 fat JAR）
JDK	OpenJDK 11

运行方式#

1
# 1. 编译打包
2
mvn clean package -DskipTests
3

4
# 2. 启动 Flink 任务
5
mvn exec:java -Dexec.mainClass="com.example.OrderFilter"
6

7
# 3. 启动 Web 服务器（另一个终端）
8
python server.py
9

10
# 4. 打开浏览器访问 http://localhost:8080

改进方向#

如果要做得更完善，有几个方向可以探索：

引入窗口机制 — 用 Flink 的滚动窗口实现定时批量统计
状态后端 — 用 RocksDB 支持大规模状态存储
数据持久化 — 把统计结果写入数据库
集群部署 — 部署到 Flink Standalone 集群实现高可用

这是大数据技术基础课程的一次实验作业。通过一个简单的订单过滤场景，走通了「浏览器 → 桥接服务 → Flink」的完整流处理链路：自定义数据源解决重连问题，零依赖的 Python 服务打通协议鸿沟，前端即时反馈提升演示体验。最大的收获是对 Flink 算子链（flatMap → filter → keyBy → sum）的执行模型有了直观认识。