MongoDB Stream是如何实现完美数据增量迁移的?

唐卓章(zale) 2018-08-08 09:20:46
作者介绍

唐卓章(zale),华为技术专家,多年互联网研发/架设经验,关注NoSQL中间件高可用及弹性扩展,在分布式系统架构性能优化方面有丰富的实践经验。目前从事物联网平台研发工作,致力于打造大容量高可用的物联网服务。

 

一、背景介绍

 

最近微服务架构火得不行,但本质上也只是风口上的一个热点词汇。


作为笔者的经验来说,想要应用一个新的架构需要带来的变革成本是非常高的。

 

尽管如此,目前还是有许多企业踏上了服务化改造的道路,这其中则免不了“旧改的各种繁杂事。


所谓的“旧改,就是把现有的系统架构来一次重构,拆分成多个细粒度的服务后,然后找时间升级割接一把,让新系统上线。这其中,数据的迁移往往会成为一个非常重要且繁杂的活儿。

 

拆分服务时数据迁移的挑战在哪?

 

  • 首先是难度大,做一个迁移方案需要了解项目的前身今世,评估迁移方案、技术工具等;

  • 其次是成本高。由于新旧系统数据结构是不一样的,需要定制开发迁移转化功能,很难有一个通用的工具能一键迁移;

  • 再者对于一些容量大、可靠性要求高的系统,要能够不影响业务,出了问题还能追溯,因此方案上还得往复杂了想。

 

二、常见方案

 

按照迁移的方案及流程,可将数据迁移分为三类:

 

1、停机迁移
 

 

最简单的方案,停机迁移的顺序如下:

 

 

采用停机迁移的好处是流程操作简单,工具成本低,然而缺点也很明显,迁移过程中业务是无法访问的,因此只适合于规格小、允许停服的场景。

 

2、业务双写
 

 

业务双写是指对现有系统先进行改造升级,支持同时对新库和旧库进行写入。之后再通过数据迁移工具对旧数据做全量迁移,待所有数据迁移转换完成后切换到新系统。

 

示意图:

 

 

业务双写的方案是平滑的,对线上业务影响极小,在出现问题的情况下可重新来过,操作压力也会比较小。

 

笔者在早些年前尝试过这样的方案,整个迁移过程确实非常顺利,但实现该方案比较复杂,需要对现有的代码进行改造并完成新数据的转换及写入,对于开发人员的要求较高。在业务逻辑清晰、团队对系统有足够的把控能力的场景下适用。

 

3、增量迁移
 

 

增量迁移的基本思路是先进行全量的迁移转换,待完成后持续进行增量数据的处理,直到数据追平后切换系统。

 

示意图:

 

 

关键点:

  • 要求系统支持增量数据的记录。对于MongoDB可以利用oplog实现这点,为避免全量迁移过程中oplog被冲掉,在开始迁移前就必须开始监听oplog,并将变更全部记录下来;如果没有办法,需要从应用层上考虑,比如为所有的表(集合)记录下updateTime这样的时间戳,或者升级应用并支持将修改操作单独记录下来。

  • 增量数据的回放是持续的。在所有的增量数据回放转换过程中,系统仍然会产生新的增量数据,这要求迁移工具能做到将增量数据持续回放并将之追平,之后才能做系统切换。

 

MongoDB 3.6版本开始便提供了Change Stream功能,支持对数据变更记录做监听。这为实现数据同步及转换处理提供了更大的便利,下面将探讨如何利用Change Stream实现数据的增量迁移。

 

三、Change Stream介绍

 

Chang Stream(变更记录流)是指collection(数据库集合)的变更事件流,应用程序通过db.collection.watch()这样的命令可以获得被监听对象的实时变更。


在该特性出现之前,你可以通过拉取oplog达到同样的目的;但oplog的处理及解析相对复杂且存在被回滚的风险,如果使用不当的话还会带来性能问题。Change Stream可以与aggregate framework结合使用,对变更集进行进一步的过滤或转换。

参考链接:https://docs.mongodb.com/manual/aggregation/

 

由于Change Stream利用了存储在oplog中的信息,因此对于单进程部署的MongoDB无法支持Change Stream功能,其只能用于启用了副本集的独立集群或分片集群。

 

监听的目标
 

 

 

变更事件
 

 

一个Change Stream Event的基本结构如下所示:

 

 

字段说明:

 

 

Change Steram支持的变更类型有以下几个:

 

 

利用以下的shell脚本,可以打印出集合 T_USER上的变更事件:

 

 

下面提供一些样例,感受一下:

 

insert事件

 

 

update事件

 

 

replace事件

 

 

delete事件

 

 

invalidate事件

 

 

更多的Change Event信息可以参考:https://docs.mongodb.com/manual/reference/change-events/

 

四、实现增量迁移

 

本次设计了一个简单的论坛帖子迁移样例,用于演示如何利用Change Stream实现完美的增量迁移方案。


背景如下:


现有的系统中有一批帖子,每个帖子都属于一个频道(channel),如下表:

 

 

新系统中频道字段将采用英文简称,同时要求能支持平滑升级。根据前面篇幅的叙述,我们将使用Change Stream功能实现一个增量迁移的方案。

 

相关表的转换如下图:

 

 

原理


topic是帖子原表,在迁移开始前将开启watch任务持续获得增量数据,并记录到 topic_incr表中;接着执行全量的迁移转换,之后再持续对增量表数据进行迁移,直到无新的增量为止。

 

接下来我们使用Java程序来完成相关代码,mongodb-java--driver在3.6版本后才支持watch功能,需要确保升级到对应版本:

 

 

定义Channel频道的转换表:

 

 

public static enum Channel {

    Food("美食"),

    Emotion("情感"),

    Pet("宠物"),

    House("家居"),

    Marriage("征婚"),

    Education("教育"),

    Travel("旅游")

    ;

    private final String oldName;

    public String getOldName() {

        return oldName;

    }

    private Channel(String oldName) {

        this.oldName = oldName;

    }

    /**

     * 转换为新的名称

     *

     * @param oldName

     * @return

     */

    public static String toNewName(String oldName) {

        for (Channel channel : values()) {

            if (channel.oldName.equalsIgnoreCase(oldName)) {

                return channel.name();

            }

        }

        return "";

    }

    /**

     * 返回一个随机频道

     *

     * @return

     */

    public static Channel random() {

        Channel[] channels = values();

        int idx = (int) (Math.random() * channels.length);

        return channels[idx];

    }

}

 

为topic表预写入1w条记录:

 

 

开启监听任务,将topic上的所有变更写入到增量表:

 

 

代码中通过watch命令获得一个MongoCursor对象,用于遍历所有的变更。


FullDocument.UPDATE_LOOKUP选项启用后,在update变更事件中将携带完整的文档数据(FullDocument)。

 

watch()命令提交后,mongos会与分片上的mongod(主节点)建立订阅通道,这可能需要花费一点时间。

 

为了模拟线上业务的真实情况,启用几个线程对topic表进行持续写操作:

 

 

ChangeTask实现逻辑如下:

 

 

每一个变更任务会不断对topic产生写操作,触发一系列ChangeEvent产生:

 

  • doInsert:生成随机频道的topic后,执行insert;

  • doUpdate:随机取得一个topic,将其channel字段改为随机值,执行update;

  • doReplace:随机取得一个topic,将其channel字段改为随机值,执行replace;

  • doDelete:随机取得一个topic,执行delete。

 

以doUpdate为例,实现代码如下:

 

 

启动一个全量迁移任务,将topic表中数据迁移到topic_new新表:

 

 

在全量迁移开始前,先获得当前时刻的的最大 _id 值(可以将此值记录下来)作为终点,随后逐个完成迁移转换。

 

在全量迁移完成后,便开始最后一步:增量迁移。

注:增量迁移过程中,变更操作仍然在进行。

 

 

final MongoCollection<Document> topicIncrCollection = getCollection(coll_topic_incr);

final MongoCollection<Document> topicNewCollection = getCollection(coll_topic_new);

ObjectId currentId = null;

Document sort = new Document("_id", 1);

MongoCursor<Document> cursor = null;

// 批量大小

int batchSize = 100;AtomicInteger count = new AtomicInteger(0);

try {

    while (true) {

        boolean isWatchTaskStillRunning = watchFlag.getCount() > 0;

        // 按ID增量分段拉取

        if (currentId == null) {

            cursor = topicIncrCollection.find().sort(sort).limit(batchSize).iterator();

        } else {

            cursor = topicIncrCollection.find(new Document("_id", new Document("$gt", currentId)))

                    .sort(sort).limit(batchSize).iterator();

        }

        boolean hasIncrRecord = false;

        while (cursor.hasNext()) {

            hasIncrRecord = true;

            Document incrDoc = cursor.next();

            OperationType opType = OperationType.fromString(incrDoc.getString(field_op));

            ObjectId docId = incrDoc.getObjectId(field_key);

            // 记录当前ID

            currentId = incrDoc.getObjectId("_id");

            if (opType == OperationType.DELETE) {

                topicNewCollection.deleteOne(new Document("_id", docId));

            } else {

                Document doc = incrDoc.get(field_data, Document.class);

                // channel转换

                String oldChannel = doc.getString(field_channel);

                doc.put(field_channel, Channel.toNewName(oldChannel));

                // 启用upsert

                UpdateOptions options = new UpdateOptions().upsert(true);

                topicNewCollection.replaceOne(new Document("_id", docId),

                        incrDoc.get(field_data, Document.class), options);

            }

            if (count.incrementAndGet() % 10 == 0) {

                logger.info("IncrTransferTask progress, count: {}", count.get());

            }

        }

        // 当watch停止工作(没有更多变更),同时也没有需要处理的记录时,跳出

        if (!isWatchTaskStillRunning && !hasIncrRecord) {

            break;

        }

        sleep(200);

    }

} catch (Exception e) {

    logger.error("IncrTransferTask ERROR", e);

}

 

增量迁移的实现是一个不断tail的过程,利用 **_id 字段的有序特性 ** 进行分段迁移;即记录下当前处理的_id值,循环拉取在该_id值之后的记录进行处理。

 

增量表(topic_incr)中除了DELETE变更之外,其余的类型都保留了整个文档,因此可直接利用replace + upsert追加到新表。

 

最后,运行整个程序。

 

 

查看topic表和topic_new表,发现两者数量是相同的。为了进一步确认一致性,我们对两个表的分别做一次聚合统计:

 

topic表

 

 

topic_new表

 

 

前者输出结果:

 

 

后者输出结果:

 

 

前后对比的结果是一致的。

 

五、后续优化

 

前面的章节演示了一个增量迁移的样例,在投入到线上运行之前,这些代码还得继续优化:

 

  • 写入性能,线上的数据量可能会达到亿级,在全量、增量迁移时应采用合理的批量化处理;另外可以通过增加并发线程,添置更多的Worker,分别对不同业务库、不同表进行处理以提升效率。增量表存在幂等性,即回放多次其最终结果还是一致的,但需要保证表级有序,即一个表同时只有一个线程在进行增量回放。

  • 容错能力,一旦watch监听任务出现异常,要能够从更早的时间点开始(使用startAtOperationTime参数),而如果写入时发生失败,要支持重试。

  • 回溯能力,做好必要的跟踪记录,比如将转换失败的ID号记录下来,旧系统的数据需要保留,以免在事后追究某个数据问题时找不着北。

  • 数据转换,新旧业务的差异不会很简单,通常需要借助大量的转换表来完成。

  • 一致性检查,需要根据业务特点开发自己的一致性检查工具,用来证明迁移后数据达到想要的一致性级别。

 

BTW,数据迁移一定要结合业务特性、架构差异来做考虑,否则还是在耍流氓。

 

六、小结

 

服务化系统中扩容、升级往往会进行数据迁移,对于业务量大,中断敏感的系统通常会采用平滑迁移的方式。


MongoDB 3.6版本后提供了Change Stream功能以支持应用订阅数据的变更事件流,本文使用Stream功能实现了增量平滑迁移的例子,这是一次尝试,相信后续这样的应用场景会越来越多。

 

附参考文档
 

 

 

作者:唐卓章(zale

源:Mongoing中文社区订阅号(ID:mongoing-mongoing)

dbaplus社群欢迎广大技术人员投稿,投稿邮箱:editor@dbaplus.cn

活动预告