利用Debezium与PubSub构建从MySQL到Weaviate及ArangoDB的实时数据管道

最初的系统架构非常清晰，甚至有些过时：一个核心的Java服务，背后是单一的MySQL实例。它处理着所有的业务逻辑——用户注册、商品管理、订单交易，以及日益增长的搜索和推荐请求。当数据量和并发量还处在可控范围时，这种单体架构凭借其简单性表现尚可。但随着业务的扩张，瓶颈开始显现。复杂的JOIN查询拖慢了核心交易，全文搜索的LIKE '%...%'查询更是灾难，而基于用户关系的推荐计算，每次都需要对巨大的关联表进行全表扫描，这在生产环境中是完全不可接受的。

问题很明确：单一的关系型数据库无法同时高效地满足OLTP、向量检索和图查询这三种截然不同的工作负载。强行让MySQL承担所有角色，只会导致所有功能都表现平庸。

初步的解决方案是引入专门的数据库。我们选择了Weaviate来处理商品描述和用户评论的语义搜索，选择了ArangoDB来构建用户关系和商品知识图谱。但这立刻引出了下一个，也是更棘手的问题：数据同步。如何在主库MySQL发生变更时，近乎实时地将数据同步到Weaviate和ArangoDB，并保证数据模型的正确转换？

最初团队有人提出“双写”方案，即在应用层代码中，完成MySQL的事务后，再同步调用Weaviate和ArangoDB的API。这个提议在架构评审会上被立刻否决。双写引入了强耦合，增加了应用层代码的复杂度和响应延迟。更致命的是，它无法保证数据一致性。如果MySQL写入成功，但写入Weaviate时失败，系统就进入了不一致状态，后续的修复和对账成本极高。

于是，我们转向了基于日志的变更数据捕获（Change Data Capture, CDC）。

架构选型：Debezium与Google Cloud Pub/Sub的组合拳

CDC的核心思想是，与其在应用层拦截数据变更，不如直接从数据库的事务日志（在MySQL中是binlog）中捕获这些变更。这种方式对源数据库是非侵入式的，性能开销极小，并且能保证捕获所有已提交的事务，不会丢失数据。

Debezium是这个领域的佼-佼者。它提供了一系列的连接器（Connectors），可以伪装成一个MySQL的从库，实时读取binlog，将INSERT、UPDATE、DELETE操作解析成结构化的JSON事件，然后推送到消息队列中。

而消息队列的选择，我们放弃了自建Kafka的方案，转向了Google Cloud Pub/Sub。在真实项目中，运维成本是重要的考量因素。Pub/Sub作为全托管服务，提供了几乎无限的伸缩能力、可靠的消息传递保证和简单的API，让我们能更专注于业务逻辑的实现，而不是基础设施的维护。

最终的架构图如下所示：

graph TD
    subgraph "源数据库"
        MySQL[(MySQL 8.0)]
    end

    subgraph "CDC & 消息总线"
        A[Kafka Connect] -- Debezium Connector --> MySQL
        A -- Pub/Sub Sink Connector --> B(Google Cloud Pub/Sub)
    end

    subgraph "消费者服务 (Python)"
        C{Data Sync Service}
    end

    subgraph "目标数据库"
        D[Weaviate]
        E[ArangoDB]
    end

    MySQL -- binlog --> A
    B -- Pull Messages --> C
    C -- 写入向量数据 --> D
    C -- 写入图/文档数据 --> E

这个架构的核心在于解耦。MySQL作为数据源，完全不知道下游有Weaviate或ArangoDB的存在。消费者服务也只关心Pub/Sub中的消息，不关心这些消息是如何产生的。这种松耦合的设计使得未来增加新的数据接收端（例如一个Elasticsearch集群或数据仓库）变得异常简单。

生产级实现细节

1. Debezium连接器配置

首先，我们需要在Kafka Connect集群上部署Debezium的MySQL连接器和Pub/Sub Sink连接器。这里的关键是Debezium连接器的配置，它决定了我们捕获哪些数据以及事件的格式。

这是一个生产环境的配置示例，包含了错误处理和元数据信息：

{
    "name": "mysql-main-source-connector",
    "config": {
        "connector.class": "io.debezium.connector.mysql.MySqlConnector",
        "tasks.max": "1",
        
        "database.hostname": "mysql-host.internal",
        "database.port": "3306",
        "database.user": "debezium_user",
        "database.password": "your_secure_password",
        "database.server.id": "184054",
        "database.server.name": "main_db_server",
        "database.include.list": "my_app_db",
        "table.include.list": "my_app_db.products,my_app_db.users,my_app_db.user_follows",

        "database.history.kafka.bootstrap.servers": "kafka-connect-internal-kafka:9092",
        "database.history.kafka.topic": "dbhistory.main_db",

        "snapshot.mode": "initial",
        "snapshot.locking.mode": "minimal",

        "key.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "key.converter.schemas.enable": "false",
        "value.converter.schemas.enable": "false",

        "tombstones.on.delete": "false",
        "include.schema.changes": "true",

        "transforms": "unwrap",
        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
        "transforms.unwrap.drop.tombstones": "false",
        "transforms.unwrap.delete.handling.mode": "rewrite"
    }
}

关键配置解析:

• database.server.id: 必须是唯一的，用于标识自己是MySQL集群中的一个“从库”。
• table.include.list: 这是个重要的优化。我们只捕获我们关心的表，避免不相关的binlog事件淹没消息总线。
• database.history.kafka.topic: Debezium用它来记录数据库schema的变更历史，这是实现无锁schema演进的关键。
• snapshot.mode: initial表示连接器首次启动时，会对所有指定的表进行一次全量快照，确保数据基线一致。对于已经存在大量数据的表，这可能会消耗很长时间，需要规划好上线窗口。
• value.converter.schemas.enable: 设置为false可以得到一个更简洁的JSON payload，否则Debezium会把schema信息也打包进去，增加消息体积。
• transforms: 这是Debezium的强大功能。我们使用ExtractNewRecordState转换，它会将复杂的Debezium事件（包含before、after、op等字段）解构成一个只包含变更后数据的简单JSON对象，这极大地简化了下游消费者的处理逻辑。对于DELETE操作，delete.handling.mode: rewrite会生成一个包含__deleted: true字段的记录，而不是一个null消息体（tombstone record），这对于下游处理也更为方便。

2. 核心消费者服务

消费者服务是整个管道的大脑。我们使用Python编写，因为它拥有成熟的云服务SDK和数据库客户端。

# main_consumer.py
import os
import json
import logging
import threading
from concurrent import futures

from google.cloud import pubsub_v1
import weaviate
from arango import ArangoClient

# --- 日志配置 ---
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(threadName)s - %(levelname)s - %(message)s'
)

# --- 全局配置与客户端初始化 ---
PROJECT_ID = os.environ.get("GCP_PROJECT_ID")
SUBSCRIPTION_ID = "cdc-mysql-sync-sub"
WEAVIATE_URL = os.environ.get("WEAVIATE_URL")
ARANGO_HOST = os.environ.get("ARANGO_HOST")
ARANGO_USER = os.environ.get("ARANGO_USER")
ARANGO_PASSWORD = os.environ.get("ARANGO_PASSWORD")
DB_NAME = "realtime_sync_db"

# Weaviate 客户端 (确保鉴权方式符合你的设置)
weaviate_client = weaviate.Client(url=WEAVIATE_URL)

# ArangoDB 客户端
arango_sys_client = ArangoClient(hosts=ARANGO_HOST)
sys_db = arango_sys_client.db('_system', username=ARANGO_USER, password=ARANGO_PASSWORD)
if not sys_db.has_database(DB_NAME):
    sys_db.create_database(DB_NAME)
db = arango_sys_client.db(DB_NAME, username=ARANGO_USER, password=ARANGO_PASSWORD)

# 确保集合存在
if not db.has_collection('products'):
    db.create_collection('products')
if not db.has_collection('users'):
    db.create_collection('users')
if not db.has_graph('user_relations'):
    graph = db.create_graph('user_relations')
    graph.create_edge_definition(
        edge_collection='user_follows',
        from_vertex_collections=['users'],
        to_vertex_collections=['users']
    )
else:
    graph = db.graph('user_relations')

products_collection = db.collection('products')
users_collection = db.collection('users')
user_follows_edge_collection = graph.edge_collection('user_follows')


def process_product_event(payload: dict):
    """处理产品表的变更事件"""
    product_id = payload.get('id')
    if not product_id:
        logging.warning("Product event missing 'id'. Payload: %s", payload)
        return

    # Debezium的DELETE操作处理
    if payload.get('__deleted') == 'true':
        try:
            # Weaviate 删除
            weaviate_client.data_object.delete(
                uuid=str(product_id), # Weaviate的UUID通常是主键
                class_name="Product"
            )
            # ArangoDB 删除
            if products_collection.has(str(product_id)):
                products_collection.delete(str(product_id))
            logging.info(f"Deleted product {product_id} from Weaviate and ArangoDB.")
        except Exception as e:
            logging.error(f"Failed to delete product {product_id}. Error: {e}", exc_info=True)
            raise  # 重新抛出异常，让消息处理失败
        return

    # 准备写入Weaviate的数据 (这里省略了调用embedding模型的步骤)
    # 在真实项目中，这里会调用一个模型服务来为 `description` 生成向量
    weaviate_obj = {
        "name": payload.get("name"),
        "description": payload.get("description"),
    }
    try:
        # Weaviate 使用 upsert 逻辑
        weaviate_client.data_object.replace(
            uuid=str(product_id),
            class_name="Product",
            data_object=weaviate_obj
        )
        
        # ArangoDB 使用 upsert 逻辑
        # 使用MySQL的`id`作为ArangoDB的`_key`来保证幂等性
        arangodb_doc = {
            '_key': str(product_id),
            'name': payload.get('name'),
            'price': payload.get('price'),
            'stock': payload.get('stock'),
            'updated_at': payload.get('updated_at')
        }
        products_collection.insert(arangodb_doc, overwrite=True)
        logging.info(f"Upserted product {product_id} to Weaviate and ArangoDB.")

    except Exception as e:
        logging.error(f"Failed to process product {product_id}. Error: {e}", exc_info=True)
        raise


def process_user_event(payload: dict):
    """处理用户表的变更事件"""
    user_id = payload.get('id')
    if not user_id:
        logging.warning("User event missing 'id'. Payload: %s", payload)
        return
        
    doc_key = str(user_id)
    if payload.get('__deleted') == 'true':
        if users_collection.has(doc_key):
            users_collection.delete(doc_key)
            logging.info(f"Deleted user {doc_key}.")
        return

    doc = {
        '_key': doc_key,
        'username': payload.get('username'),
        'registration_date': payload.get('registration_date')
    }
    users_collection.insert(doc, overwrite=True)
    logging.info(f"Upserted user {doc_key}.")

def process_user_follows_event(payload: dict):
    """处理用户关注关系表的变更事件"""
    follower_id = payload.get('follower_id')
    following_id = payload.get('following_id')

    if not follower_id or not following_id:
        logging.warning("Follow event missing IDs. Payload: %s", payload)
        return

    # 边(edge)的_key通常由源和目标顶点组合而成，保证唯一性和幂等性
    edge_key = f"{follower_id}-{following_id}"
    from_vertex = f"users/{follower_id}"
    to_vertex = f"users/{following_id}"

    if payload.get('__deleted') == 'true':
        if user_follows_edge_collection.has(edge_key):
            user_follows_edge_collection.delete(edge_key)
            logging.info(f"Deleted follow edge: {follower_id} -> {following_id}.")
        return
    
    edge_data = {
        '_key': edge_key,
        '_from': from_vertex,
        '_to': to_vertex,
        'created_at': payload.get('created_at')
    }
    user_follows_edge_collection.insert(edge_data, overwrite=True)
    logging.info(f"Upserted follow edge: {follower_id} -> {following_id}.")


def callback(message: pubsub_v1.subscriber.message.Message):
    """Pub/Sub消息处理回调"""
    try:
        # Debezium sink到Pub/Sub时，key是表名，value是行数据
        table_name = message.attributes.get("debezium.source.table")
        data = message.data.decode("utf-8")
        payload = json.loads(data)
        
        logging.info(f"Received message for table: {table_name}")

        if table_name == 'products':
            process_product_event(payload)
        elif table_name == 'users':
            process_user_event(payload)
        elif table_name == 'user_follows':
            process_user_follows_event(payload)
        else:
            logging.warning(f"No handler for table: {table_name}")

        message.ack()
    except json.JSONDecodeError as e:
        logging.error(f"Failed to decode JSON message: {message.data}. Error: {e}")
        # 对于格式错误的消息，直接ack掉，避免无限重试，或者推送到死信队列
        message.ack()
    except Exception as e:
        logging.critical(f"Unhandled exception during message processing. Message will be nacked. Error: {e}", exc_info=True)
        message.nack() # 让Pub/Sub重新投递消息

def main():
    """主函数，启动消费者"""
    subscriber = pubsub_v1.SubscriberClient()
    subscription_path = subscriber.subscription_path(PROJECT_ID, SUBSCRIPTION_ID)

    # 使用线程池来并发处理消息
    # 这里的流控设置非常重要，可以防止消费者被大量消息冲垮
    flow_control = pubsub_v1.types.FlowControl(max_messages=100, max_bytes=10 * 1024 * 1024)
    executor = futures.ThreadPoolExecutor(max_workers=10)
    
    streaming_pull_future = subscriber.subscribe(
        subscription_path, 
        callback=callback,
        flow_control=flow_control,
        scheduler=pubsub_v1.subscriber.scheduler.ThreadScheduler(executor=executor)
    )

    logging.info(f"Listening for messages on {subscription_path}...")
    try:
        # block indefinitely
        streaming_pull_future.result()
    except KeyboardInterrupt:
        streaming_pull_future.cancel()
        streaming_pull_future.result()
        executor.shutdown()
    except Exception as e:
        logging.error(f"Pulling messages failed: {e}", exc_info=True)
        streaming_pull_future.cancel()
        streaming_pull_future.result()
        executor.shutdown()

if __name__ == "__main__":
    main()

单元测试思路:
对这段代码进行单元测试，不应该直接连接真实的数据库或Pub/Sub。核心思路是模拟（Mock）外部依赖。

1. 测试process_*_event函数:
   • 为每个函数准备INSERT, UPDATE, DELETE三种场景的Debezium payload（JSON字典）。
   • Mock weaviate_client 和 db.collection(...) 对象。
   • 调用process_product_event(mock_payload)。
   • 断言（Assert）被mock的客户端的replace, delete, insert等方法是否被以正确的参数调用。例如，mock_products_collection.insert.assert_called_with({'_key': '123', ...}, overwrite=True)。
2. 测试callback函数:
   • 创建一个Mock的Message对象，包含data（编码后的JSON）和attributes（表名）。
   • Mock process_product_event, process_user_event等函数。
   • 调用callback(mock_message)。
   • 断言正确的处理函数被调用，并且mock_message.ack()被调用。
   • 测试异常情况：传入一个非法的JSON字符串，断言ack()被调用；让处理函数抛出异常，断言nack()被调用。

生产中的陷阱与考量

这个架构看似完美，但在真实部署中我们还是遇到了一些问题。

1. 幂等性是生命线: Pub/Sub保证至少一次（At-Least-Once）的消息传递。这意味着同一个消息可能被重复投递。我们的消费者必须是幂等的。在上面的代码中，我们通过将MySQL的主键id作为Weaviate的UUID和ArangoDB的_key来实现幂等性。无论消息被处理多少次，结果都是一样的。对于user_follows这样的关联表，复合主键（follower_id, following_id）的组合可以作为_key。

2. Schema演进: 当MySQL中的表结构发生变化时，例如products表增加一个tags字段，会发生什么？Debezium能够捕获ALTER TABLE事件，并将其记录到database.history主题中。下游的消费者必须具备足够的鲁棒性来处理payload中新增或减少的字段。在Python中，payload.get("new_field")返回None而不是抛出KeyError，这提供了一定程度的向前兼容。但对于破坏性变更（如字段重命名或删除），则需要更周密的发布计划，可能需要先更新消费者代码，再执行数据库变更。

3. 全量快照的冲击: 在一个拥有数亿行记录的大表上启动Debezium的initial快照，会对源数据库产生巨大的读压力，并且会向Pub/Sub瞬间推送海量消息。这可能会打垮配置不足的消费者实例。一个常见的策略是：在低峰期启动快照，并为消费者配置好自动伸缩（例如在GKE上使用HPA），同时设置严格的Pub/Sub流控参数（max_messages）来控制消费速度。

这个基于CDC的实时数据管道，最终成功地将我们的单体数据库压力分解到了多个专门的、高性能的查询系统中。它不仅解决了性能瓶颈，其松耦合的特性也为未来的系统演进提供了极大的灵活性。

然而，这套系统的复杂性也不容忽视。它引入了多个新的组件，对可观测性（监控Debezium延迟、Pub/Sub积压、消费者错误率）提出了更高的要求。数据一致性模型也从强一致性变为了最终一致性，业务端需要能够接受秒级的同步延迟。对于需要跨越多个数据库进行原子性写入的场景，这套架构并不适用，那时可能需要引入Saga等分布式事务模式。技术的选型总是在不同维度间的权衡，没有银弹，只有最适合当前业务场景的解决方案。