如何使用Pathway和Debezium实现MongoDB实时数据处理：完整指南

如何使用Pathway和Debezium实现MongoDB实时数据处理：完整指南

【免费下载链接】pathwayPathway is an open framework for high-throughput and low-latency real-time data processing.项目地址: https://gitcode.com/GitHub_Trending/pa/pathway

Pathway是一个开源框架，专为高吞吐量和低延迟的实时数据处理而设计。本文将详细介绍如何利用Pathway的Debezium连接器实现MongoDB数据库的实时变更捕获（CDC），帮助新手用户轻松搭建实时数据处理管道。

为什么选择Pathway进行MongoDB实时处理？

传统数据库如MongoDB并非为流处理场景设计，因此需要变更数据捕获（CDC）机制来监控数据库变化并生成数据流。Pathway提供了与Debezium的无缝集成，通过pw.io.debezium.read连接器捕获MongoDB变更，并使用pw.io.mongodb.write将处理结果写回数据库，实现端到端的实时数据处理。

Pathway实时监控仪表板展示内存使用、延迟和CPU时间等关键指标

快速入门：5分钟实现实时数据求和

想象一个简单场景：MongoDB中有一个values集合，需要实时计算其中value字段的总和并存储到sum_values集合。使用Pathway只需几行代码即可实现：

import pathway as pw # Kafka连接配置 input_rdkafka_settings = { "bootstrap.servers": "kafka:9092", "security.protocol": "plaintext", "group.id": "0", "auto.offset.reset": "earliest", } class InputSchema(pw.Schema): value: int # 从Debezium读取MongoDB变更流 t = pw.io.debezium.read( input_rdkafka_settings, topic_name="my_mongo_db.test_database.values", schema=InputSchema, autocommit_duration_ms=100, ) # 实时计算总和 t = t.reduce(sum=pw.reducers.sum(t.value)) # 结果写回MongoDB pw.io.mongodb.write( t, connection_string="mongodb://mongodb:27017/?replicaSet=rs0", database="my_mongo_db", collection="sum_values", ) pw.run()

完整架构：Pathway + Debezium + MongoDB

实现MongoDB实时处理需要以下组件协同工作：

1. MongoDB：存储原始数据和处理结果
2. Debezium：捕获MongoDB变更并转换为Kafka消息
3. Kafka/ZooKeeper：消息队列，传递变更事件
4. Pathway：处理数据流并写回结果

在Jupyter Notebook中使用Pathway处理实时数据流

步骤1：配置MongoDB副本集

Debezium需要MongoDB副本集来捕获变更，通过Docker Compose配置：

mongodb: image: mongo command: ["--replSet", "rs0", "--port", "27017"] healthcheck: test: echo "try { rs.status() } catch (err) { rs.initiate({_id:'rs0',members:[{_id:0,host:'mongodb:27017'}]}) }" | mongosh --port 27017 --quiet interval: 5s timeout: 120s

步骤2：设置Debezium连接器

创建Debezium配置脚本connector.sh，连接MongoDB和Kafka：

curl -H 'Content-Type: application/json' debezium:8083/connectors --data '{ "name": "values-connector", "config": { "connector.class": "io.debezium.connector.mongodb.MongoDbConnector", "mongodb.hosts": "rs0/mongodb:27017", "mongodb.name": "my_mongo_db", "database.include.list": "test_database", "database.history.kafka.bootstrap.servers": "kafka:9092", "database.history.kafka.topic": "dbhistory.mongo" } }'

步骤3：使用Pathway处理数据流

完整代码结构位于examples/projects/debezium-mongodb-example，核心处理逻辑在sum.py中实现：

• 使用pw.io.debezium.read从Kafka读取Debezium消息
• 定义数据模式InputSchema匹配MongoDB文档结构
• 通过reduce操作计算实时总和
• 使用pw.io.mongodb.write将结果写回数据库

运行与监控

通过Makefile简化部署流程：

build: chmod +x ./debezium/connector.sh docker compose up -d docker compose exec debezium ./connector.sh stop: docker compose down -v

执行make启动所有服务，通过以下命令查看实时计算结果：

docker-compose exec mongodb mongosh use my_mongo_db db["sum_values"].find().sort({ time: -1 }).pretty()

总结

Pathway提供了简单而强大的工具集，使MongoDB实时数据处理变得轻松。通过Debezium连接器，您可以捕获数据库变更并构建实时处理管道，而无需复杂的基础设施配置。无论是实时分析、监控还是数据同步，Pathway都能提供高性能和低延迟的解决方案。

更多详细文档请参考官方指南，完整示例代码可在项目仓库中找到。立即尝试使用Pathway，开启您的实时数据处理之旅！

【免费下载链接】pathwayPathway is an open framework for high-throughput and low-latency real-time data processing.项目地址: https://gitcode.com/GitHub_Trending/pa/pathway