技术内幕｜使用 Debezium 和 RisingWave 对 MongoDB 进行持续分析

1. MongoDB 和流式 Join 的挑战

作为一种操作型数据库，MongoDB 在提供快速数据操作和查询性能方面表现十分出色。然而，在维护实时视图或执行流处理任务的内置支持方面，它确实存在一些局限性。例如，MongoDB 不支持连接两个集合并实时刷新结果（尤其是高频率按秒刷新）。

在 MongoDB 中，Join 使用的是 $lookup 聚合算子，而非传统的 SQL 式 Join：

db.users.aggregate([{
  $lookup: {
    from: "products",
    localField: "product_id",
    foreignField: "_id",
    as: "products"
  }
}])

当前的方法虽然有效，但在连接两个以上的集合时会比较繁琐和不便。

MongoDB 提倡去规范化，如果有最佳的 Schema，很多情况下就可以避免使用 Join。但是，实际情况可能会比较混乱，因此有时需要使用 Join。与其依赖复杂的 MongoDB 聚合，不如将此任务委托给 RisingWave 这样的专用流处理系统。

2. RisingWave 解决方案

一个实用的解决方案是通过 Kafka 将 MongoDB 变更流导入到 RisingWave，从而实现灵活的实时连接。

RisingWave 可充当实时数据源的中心枢纽。来自 MongoDB 和 Kafka 的数据可以导入到 RisingWave 并进行连接。RisingWave 的状态后端 Hummock 利用云对象存储，提供了弹性和充足的容量。这能够支持在多个 Source 上执行大型 Join，处理 10 个以上的多路 Join。

在这个数据栈中，Debezium 起着至关重要的作用。它提取 MongoDB oplog 条目并将其导出到 Kafka Topic，然后由 RisingWave 消费。例如：

CREATE TABLE source_name (
   _id jsonb PRIMARY KEY,
   payload jsonb
)
WITH (
   connector='kafka',
   topic='debezium_mongo_json_customers',
   properties.bootstrap.server='172.10.1.1:9090,172.10.1.2:9090',
   scan.startup.mode = 'earliest'
) FORMAT DEBEZIUM_MONGO ENCODE JSON;

创建了 MongoDB Source 之后，我们就可以创建物化视图，实时连接来自 MongoDB 和其他 Source 的数据。

有关此过程的详细步骤，请参阅 RisingWave 文档。

3. 处理 MongoDB 的 JSON 数据

RisingWave 可以利用从 PostgreSQL 继承的 JSONB 支持，分析 MongoDB 中的 JSON（BSON）数据。这样，即使在非关系型的 Schema 中也能直接进行摄取和查询。虽然 MongoDB 的原生 API 在数据操作方面表现十分出色，在 Node.js 网络应用程序中尤其如此，但进行分析性工作负载需要不同的专长。SQL 仍是最流行的数据分析语言。目前，RisingWave 提供了超过 30 个 JSON 函数，包括对 JSONPath 的支持。这使得用户在转换 MongoDB 数据时无需编写 UDF。

以下是在 RisingWave 中处理 JSONB 数据的一些示例：

查找年龄在 25 至 30 岁之间的用户：

SELECT *
FROM users
WHERE (payload->>'age')::int BETWEEN 25 AND 30;

通过姓名和电子邮件查找用户：

SELECT *
FROM users
WHERE payload @> '{"name": "Bob"}'
AND payload->>'email' LIKE 'bob@example.com';

4. 结论

RisingWave 能够很好地解析由 Debezium 提取的 MongoDB 变更流。它的云原生存储使其可连接多个 MongoDB 集合，并创建可被其他服务使用的统一流。同时，RisingWave 支持 JSONB，可轻松处理 MongoDB 文档，为实时数据处理挑战提供了强大的解决方案。

关于 RisingWave