Skip to content

增量抽取数据失败,使用change_stream模式增量抽取只有插入操作的时序表,一直是 write_success=0 。如果incr_sync.mongo_fetch_method转成用oplog来做增量抽取又会报错误。 #1000

Open

Description

@harzw

MongoShake的版本: mongo-shake-v2.8.8

源端MongoDB 版本: 7.0.28
目标端MongoDB版本:阿里云MongoDB 8.0.18-20260812202609 (看文档对应基准版本是7.0.39)

时序表:operation.gpsevents_20260X

**change_stream 模式下日志一直是 write_success=0 ,但是check point又一直往前推进 **

[2026/09/08 13:35:30 CST] [INFO] [name=ciotrep, stage=incr, get=754333, filter=754314, write_success=0, tps=0, ckpt_times=5, lsn_ckpt={7683033143435067406[1788845552, 14], 2026-09-08 13:32:32}, lsn_ack={0[0, 0], 1970-01-01 08:00:00}]]
[2026/09/08 13:35:33 CST] [INFO] Syncer[ciotrep] try to update checkpoint mandatory from 7683033143435067406[1788845552, 14] to {1788845733 35}
[2026/09/08 13:35:33 CST] [INFO] Syncer[ciotrep] waiting last checkpoint[7683033920824148003[1788845733, 35]] updated
[2026/09/08 13:35:33 CST] [INFO] Syncer[ciotrep] last checkpoint[7683033920824148003[1788845733, 35]] updated [false]
[2026/09/08 13:35:33 CST] [INFO] checkpoint update sync.ckptTime to:2026-09-08 13:35:33.305656873 +0800 CST m=+1094.444924388
[2026/09/08 13:35:33 CST] [INFO] checkpoint func lowest:7683033920824148003[1788845733, 35] inMemoryTs:7683033143435067406[1788845552, 14] flush:false inputTs:7683033920824148003
[2026/09/08 13:35:33 CST] [INFO] ciotrep Record new checkpoint in MongoDB success [1788845733]
[2026/09/08 13:35:33 CST] [INFO] CheckpointOperation write success. updated from 7683033143435067406[1788845552, 14] to 7683033920824148003[1788845733, 35]
[2026/09/08 13:35:35 CST] [INFO] [name=ciotrep, stage=incr, get=755946, filter=755914, write_success=0, tps=0, ckpt_times=6, lsn_ckpt={7683033920824148003[1788845733, 35], 2026-09-08 13:35:33}, lsn_ack={0[0, 0], 1970-01-01 08:00:00}]]
[2026/09/08 13:35:40 CST] [INFO] [name=ciotrep, stage=incr, get=757245, filter=757194, write_success=0, tps=0, ckpt_times=6, lsn_ckpt={7683033920824148003[1788845733, 35], 2026-09-08 13:35:33}, lsn_ack={0[0, 0], 1970-01-01 08:00:00}]]
[2026/09/08 13:35:45 CST] [INFO] [name=ciotrep, stage=incr, get=758413, filter=758402, write_success=0, tps=0, ckpt_times=6, lsn_ckpt={7683033920824148003[1788845733, 35], 2026-09-08 13:35:33}, lsn_ack={0[0, 0], 1970-01-01 08:00:00}]]
[2026/09/08 13:35:50 CST] [INFO] [name=ciotrep, stage=incr, get=760053, filter=760017, write_success=0, tps=0, ckpt_times=6, lsn_ckpt={7683033920824148003[1788845733, 35], 2026-09-08 13:35:33}, lsn_ack={0[0, 0], 1970-01-01 08:00:00}]]

oplog 模式下报错日志如下

[2026/09/08 13:37:54 CST] [INFO] Close client with mongodb://root:XXX@XXX:3717,XXX:3717
[2026/09/08 13:37:54 CST] [INFO] update or.query to map[ts:map[$gt:{1788845746 252}]]
[2026/09/08 13:37:54 CST] [INFO] New session to mongodb://root:XXX@XXX:3717,XXX:3717 successfully
[2026/09/08 13:37:54 CST] [WARNING] insert docs with length[1] into ns[operation.system.buckets.gpsevents_202608] of dest mongo failed[bulk write exception: write errors: [bypassDocumentValidation is not permitted on timeseries collections]] res[&{0 0 0 0 0 map[]}]
[2026/09/08 13:37:54 CST] [CRITICAL] Replay-3, executor-3, oplog for namespace[operation.system.buckets.gpsevents_202608] op[i] failed. error type[mongo.BulkWriteException] error[bulk write exception: write errors: [bypassDocumentValidation is not permitted on timeseries collections]], logs number[1], firstLog: {"ts":{"T":1788845738,"I":116},"t":52,"h":null,"v":2,"op":"i","ns":"operation.system.buckets.gpsevents_202608","o":[{"Key":"_id","Value":"6a91be908f5ed5c1c8e58a80"},{"Key":"control","Value":[{"Key":"version","Value":1},{"Key":"min","Value":[{"Key":"_id","Value":"6a9f9eaaf8b2b632bad49f7f"},{"Key":"seqno","Value":"3540"},{"Key":"revTime","Value":"2026-09-08T13:35:36+08:00"},{"Key":"cltTime","Value":"2026-08-29T01:02:39+08:00"},{"Key":"cltDateTime","Value":"2026-08-28T17:00:00Z"}......

使用配置如下( 改变部分只有 incr_sync.mongo_fetch_method = change_stream or incr_sync.mongo_fetch_method = oplog )

# if you have any problem, please visit https://github.com/alibaba/MongoShake/wiki/FAQ
# for the detail explanation, please visit https://github.com/alibaba/MongoShake/wiki/%E9%85%8D%E7%BD%AE%E5%8F%82%E6%95%B0%E8%AF%B4%E6%98%8E
# 如果有问题,请先查看FAQ文档以及wiki上的说明。
# 关于各个参数的详细说明,请参考:wiki-配置参数说明

# current configuration version, do not modify.
# 当前配置文件的版本号,请不要修改该值。
conf.version = 14

# --------------------------- global configuration ---------------------------
# collector name
# id用于输出pid文件等信息。
id = mongoshake_timeseries

# high availability option.
# enable master election if set true. only one mongoshake can become master
# and do sync, the others will wait and at most one of them become master once 
# previous master die. The master information stores in the `mongoshake` db in the source 
# database by default.
# This option is useless when there is only one mongoshake running.
# 如果开启主备mongoshake拉取同一个源端,此参数需要开启。
master_quorum = false
# master_quorum only supports checkpoint.storage=database. When master_quorum=true,
# master_quorum.election_id is required and should be unique per independent sync job or HA group.
# master_quorum 仅支持 checkpoint.storage=database;开启后必须显式配置 master_quorum.election_id,
# 且每个独立同步任务或 HA 组应使用不同的 election id。
# master_quorum.election_id = 66e3ffdd9df1af8e2308fb53

# http api interface. Users can use this api to monitor mongoshake.
# `curl 127.0.0.1:9100`.
# We also provide a restful tool named "mongoshake-stat" to
# print ack, lsn, checkpoint and qps information based on this api.
# usage: `./mongoshake-stat --port=9100`
# Set the port to a value <= 0 to disable the corresponding listener.
# 全量和增量的restful监控端口,可以用curl查看内部监控metric统计情况;端口<=0表示关闭对应监听。
full_sync.http_port = 9101
incr_sync.http_port = 9100
# Prometheus metrics endpoint. Set the port to a value <= 0 to disable it.
# Prometheus 监控端口,访问 /metrics 获取指标;端口<=0表示关闭对应监听。
prom.http_port = 9102
# profiling on net/http/profile
# profiling端口,用于查看内部go堆栈。
system_profile_port = 9200

# global log level: debug, info, warning, error. lower level message will be filter
log.level = info
# log directory. log and pid file will be stored into this file.
# if not set, default is "./logs/"
# log和pid文件的目录,如果不设置默认打到当前路径的logs目录。
log.dir = logs
# log file name.
# log文件名。
log.file = collector_27017.log
# log flush enable. If set false, logs may not be print when exit. If
# set true, performance will be decreased extremely
# 设置log刷新,false表示包含缓存,如果true那么每条log都会直接刷屏,但对性能有影响;
# 反之,退出不一定能打印所有的log,调试时建议配置true。
log.flush = false
# log rotate when file size reaches this value in megabytes.
# 日志文件达到指定大小(MB)时轮转。
log.max_size_mb = 20
# keep rotated log files for this many days.
# 历史日志文件按天数保留。
log.max_age = 7

# sync mode: all/full/incr. default is incr.
# all means full synchronization + incremental synchronization.
# full means full synchronization only.
# incr means incremental synchronization only.
# 同步模式,all表示全量+增量同步,full表示全量同步,incr表示增量同步。
#sync_mode = all
sync_mode = incr

# connect source mongodb, set username and password if enable authority. Please note: password shouldn't contain '@'.
# split by comma(,) if use multiple instance in one replica-set. E.g., mongodb://username1:password1@primaryA,secondaryB,secondaryC
# split by semicolon(;) if sharding enable. E.g., mongodb://username1:password1@primaryA,secondaryB,secondaryC;mongodb://username2:password2@primaryX,secondaryY,secondaryZ
# 源MongoDB连接串信息,逗号分隔同一个副本集内的结点,分号分隔分片sharding实例,免密模式可以忽略“username:password@”
# 注意,密码里面如果含有'@'等特殊字符的话,shake会按照如下的URL编码方式进行处理。
# https://help.aliyun.com/zh/mongodb/support/connection-access-and-network
# 举例:
# 副本集:mongodb://username1:password1@primaryA,secondaryB,secondaryC
# 分片集:mongodb://username1:password1@primaryA,secondaryB,secondaryC;mongodb://username2:password2@primaryX,secondaryY,secondaryZ
mongo_urls = mongodb://XXXX:XXXX@XXXX:27017,XXXX:27017,XXXX:27017
# please fill the source config server url if source mongodb is sharding.
mongo_cs_url =
# please give at least one mongos address if source is sharding.
# 如果源端采用change stream拉取,这里还需要配置至少一个mongos的地址,多个mongos地址以逗号(,)分割
mongo_s_url = 
# enable source ssl
mongo_ssl_root_ca_file = 

# tunnel pipeline type. now we support rpc,file,kafka,mock,direct
# 通道模式。
tunnel = direct
# tunnel target resource url
# for rpc. this is remote receiver socket address
# for tcp. this is remote receiver socket address
# for file. this is the file path, for instance "data"
# for kafka. this is the topic and brokers address which split by comma, for
# instance: topic@brokers1,brokers2, default topic is "mongoshake"
# for mock. this is useless
# for direct. this is target mongodb address which format is the same as `mongo_urls`. If
# the target is sharding, this should be the mongos address.
# direct模式用于直接写入MongoDB,其余模式用于一些分析,或者远距离传输场景,
# 注意,如果是非direct模式,需要通过receiver进行解析,具体参考FAQ文档。
# 此处配置通道的地址,格式与mongo_urls对齐。
tunnel.address = mongodb://root:XXX@XXX:3717,XXX:3717
# the message format in the tunnel, used when tunnel is kafka.
# "raw": batched raw data format which has good performance but encoded so that users
# should parse it by receiver.
# "json": single oplog format by json.
# "bson": single oplog format by bson.
# 通道数据的类型,只用于kafka和file通道类型。
# raw是默认的类型,其采用聚合的模式进行写入和
# 读取,但是由于携带了一些控制信息,所以需要专门用receiver进行解析。
# json以json的格式写入kafka,便于用户直接读取。
# bson以bson二进制的格式写入kafka。
tunnel.message = raw
# how many partitions will be written, use some hash function in "incr_sync.shard_key".
# 如果目的端是kafka,最多启用多少个partition,最大不超过"incr_sync.worker"。默认1
tunnel.kafka.partition_number = 1
# Whether or not to use SASL authentication when connecting to the broker
# (defaults to false).
tunnel.kafka.sasl.enable = false
# for kafka. this is the username and password auth which split by comma, for instance: username@password
# UserName is the authentication identity (authcid) to present for
# SASL/PLAIN or SASL/SCRAM authentication
# Password for SASL/PLAIN authentication
tunnel.kafka.sasl.auth =
# SASLMechanism is the name of the enabled SASL mechanism.
# Currently only supported values: PLAIN、SCRAM-SHA-256和SCRAM-SHA-512 (defaults to PLAIN).
tunnel.kafka.sasl.mechanism = PLAIN
#represents the various compression codecs recognized by Kafka in messages.
#supported values:none,gzip,snappy,lz4,zstd
tunnel.kafka.compression = none
#The maximum permitted size of a message (defaults to 18*1024*1024). Should be
#set equal to or smaller than the broker's `message.max.bytes`.
#tunnel.kafka.producer.max_message_bytes = 18*1024*1024
tunnel.kafka.producer.max_message_bytes = 18874368

# tunnel json format, it'll only take effect in the case of tunnel.message = json
# and tunnel == kafka. Set canonical_extended_json if you want to use "Canonical
# Extended JSON Format", #559.
# 写入异构通道的json格式。如果希望使用Canonical Extended Json Format,则设置为
# canonical_extended_json
tunnel.json.format =
# if tunnel == direct or kafka and enable ssl
tunnel.mongo_ssl_root_ca_file =

# connect mode:
# primary: fetch data from primary.
# secondaryPreferred: fetch data from secondary if has, otherwise primary.(default)
# standalone: fetch data from given 1 node, no matter primary, secondary or hidden. This is only
# support when tunnel type is direct.
# 连接模式,primary表示从主上拉取,secondaryPreferred表示优先从secondary拉取(默认建议值),
# standalone表示从任意单个结点拉取。
mongo_connect_mode = secondaryPreferred

# filter db or collection namespace. at most one of these two parameters can be given.
# if the filter.namespace.black is not empty, the given namespace will be
# filtered while others namespace passed. 
# if the filter.namespace.white is not empty, the given namespace will be
# passed while others filtered. 
# all the namespace will be passed if no condition given.
# db and collection connected by the dot(.).
# different namespaces are split by the semicolon(;).
# filter: filterDbName1.filterCollectionName1;filterDbName2
# 黑白名单过滤,目前不支持正则,白名单表示通过的namespace,黑名单表示过滤的namespace,
# 不能同时指定。分号分割不同namespace,每个namespace可以是db,也可以是db.collection。
filter.namespace.black = 
filter.namespace.white = operation.gpsevents_202606;operation.gpsevents_202607;operation.gpsevents_202608;operation.gpsevents_202609;dry_operation.gpsevents_202306;dry_operation.gpsevents_202307;dry_operation.gpsevents_202308;dry_operation.gpsevents_202309;dry_operation.gpsevents_202310;dry_operation.gpsevents_202311;dry_operation.gpsevents_202312;dry_operation.gpsevents_202401;dry_operation.gpsevents_202402;dry_operation.gpsevents_202403;dry_operation.gpsevents_202404;dry_operation.gpsevents_202405;dry_operation.gpsevents_202406;dry_operation.gpsevents_202407;dry_operation.gpsevents_202408;dry_operation.gpsevents_202409;dry_operation.gpsevents_202410;dry_operation.gpsevents_202411;dry_operation.gpsevents_202412;dry_operation.gpsevents_202501;dry_operation.gpsevents_202502;dry_operation.gpsevents_202503;dry_operation.gpsevents_202504;dry_operation.gpsevents_202505;dry_operation.gpsevents_202506;dry_operation.gpsevents_202507;dry_operation.gpsevents_202508;dry_operation.gpsevents_202509;dry_operation.gpsevents_202510;dry_operation.gpsevents_202511;dry_operation.gpsevents_202512;dry_operation.gpsevents_202601;dry_operation.gpsevents_202602;dry_operation.gpsevents_202603;dry_operation.gpsevents_202604;dry_operation.gpsevents_202605;dry_operation.gpsevents_202606;dry_operation.gpsevents_202607;dry_operation.gpsevents_202608;dry_operation.gpsevents_202609;dry_operation.gpsevents_202610;dry_operation.gpsevents_202611;dry_operation.gpsevents_202612
# filter oplog by oplog op type. different ops are split by semicolon(;).
# valid values are "i", "u", "d", "c". keep empty to pass all oplogs.
# this setting matches oplog op types instead of only MongoDB command oplogs.
# for "i", "u", and "d", matching inner ops inside applyOps are also filtered.
# e.g. set "d" to drop delete oplogs such as TTL delete or batched delete from incremental sync.
# e.g. set "i;u" to drop matching insert/update ops even when they are nested inside applyOps.
# 按 oplog op 类型过滤。多个 op 以分号(;)分隔。合法值包括 "i"、"u"、"d"、"c"。
# 该配置匹配的是 oplog op 类型,不是只匹配 MongoDB command oplog。保持为空表示不过滤;
# 对于 "i"、"u"、"d",会同时过滤顶层 oplog 和 applyOps 内层命中的 DML 操作;
# 例如配置为 d 可过滤增量同步中的 delete oplog(如 TTL 删除、batched delete);
# 配置为 i;u 时,也会过滤 applyOps 内层命中的 insert/update。
filter.op_types =

# some databases like "admin", "local", "mongoshake", "config" are
# filtered, users can enable these database based on some special needs.
# different database are split by the semicolon(;).
# e.g., admin;mongoshake
# pay attention: collection isn't support like "admin.xxx"
# 正常情况下,不建议配置该参数,但对于有些非常特殊的场景,用户可以启用admin,mongoshake等库的同步,
# 以分号分割,例如:admin;mongoshake
filter.pass.special.db =
# only transfer oplog commands for syncing. represent
# by oplog.op are "i","d","u".
# DDL will be transferred if disable like create index, drop database,
# transaction in mongodb 4.0.
# 是否需要开启DDL同步,true表示开启,源是sharding暂时不支持开启。
# 如果目的端是sharding,暂时不支持applyOps命令,包括事务。
filter.ddl_enable = true
# filter oplog gid if enabled.
# 如果MongoDB启用了gid,但是目的端MongoDB不支持gid导致同步会失败,可以启用gid过滤,将会去掉gid字段。
# 谨慎建议开启,shake本身性能受损很大。
filter.oplog.gids = false

# checkpoint info, used in resuming from break point.
# checkpoint存储信息,用于支持断点续传。
# context.storage.url is used to mark the checkpoint store database. E.g., mongodb://127.0.0.1:20070
# if not set, checkpoint will be written into source mongodb(db=mongoshake)
# checkpoint的具体写入的MongoDB地址,如果不配置,对于副本集和分片集群都将写入源库(db=mongoshake)
# 2.4版本以后不需要配置为源端cs的地址。
checkpoint.storage.url = mongodb://root:XXX@XXX:3717,XXX:3717
# checkpoint db's name.
# checkpoint存储的db的名字
checkpoint.storage.db = mongoshake_timeseries
# checkpoint collection's name.
# checkpoint存储的表的名字,如果启动多个mongoshake拉取同一个源可以修改这个表名以防止冲突。
checkpoint.storage.collection = ckpt_default
# set if enable ssl
checkpoint.storage.url.mongo_ssl_root_ca_file = 
# real checkpoint: the fetching oplog position.
# pay attention: this is UTC time which is 8 hours latter than CST time. this
# variable will only be used when checkpoint is not exist.
# 本次开始拉取的位置,如果checkpoint已经存在(位于上述存储位置)则该参数无效,
# 如果需要强制该位置开始拉取,需要先删除原来的checkpoint,详见FAQ。
# 若checkpoint不存在,且该值为1970-01-01T00:00:00Z,则会拉取源端现有的所有oplog。
# 若checkpoint不存在,且该值不为1970-01-01T00:00:00Z,则会先检查源端oplog最老的时间是否
# 大于给定的时间,如果是则会直接报错退出。
#checkpoint.start_position = 1970-01-01T00:00:00Z
checkpoint.start_position = 2026-09-07T00:00:00Z

# transform from source db or collection namespace to dest db or collection namespace.
# at most one of these two parameters can be given.
# transform: fromDbName1.fromCollectionName1:toDbName1.toCollectionName1;fromDbName2:toDbName2
# 转换命名空间,比如a.b同步后变成c.d,谨慎建议开启,比较耗性能。
transform.namespace =

# --------------------------- full sync configuration ---------------------------
# the number of collection concurrence
# 并发最大拉取的表个数,例如,6表示同一时刻shake最多拉取6个表。
full_sync.reader.collection_parallel = 6
# the number of document writer thread in each collection.
# 同一个表内并发写的线程数,例如,8表示对于同一个表,将会有8个写线程进行并发写入。
full_sync.reader.write_document_parallel = 12
# number of documents in a batch insert in a document concurrence
# 目的端写入的batch大小,例如,128表示一个线程将会一次聚合128个文档然后再写入。
full_sync.reader.document_batch_size = 128
# number of documents in a batch in fetch from source db
# 源端拉取batch中最大条数
full_sync.reader.fetch_batch_size = 1024
# max number of fetching thread per table. default is 1
# 单个表最大拉取的线程数,默认是单线程拉取。需要具备splitVector权限。
# 注意:对单个表来说,仅支持索引对应的value是同种类型,如果有不同类型请勿启用该配置项!
full_sync.reader.parallel_thread = 1
# the parallel query index if set full_sync.reader.parallel_thread. index should only has
# 1 field.
# 如果设置了full_sync.reader.parallel_thread,还需要设置该参数,并行拉取所扫描的index,value
# 必须是同种类型。对于副本集,建议设置_id;对于集群版,建议设置shard_key。key只能有1个field。
full_sync.reader.parallel_index = _id
#full_sync.reader.parallel_index = cltDateTime
# maxChunkSize for splitVector, default is 1024MB, this value should be [1,1024]MB since below versions:
# 4.2.24, 4.4.19, 5.0.13, 6.0.2, see more details in: https://github.com/alibaba/MongoShake/issues/923
# 要使用单表并发拉取,splitVector命令的maxChunkSize从某些小版本开始就不能超过1024MB,否则会报错。(4.2.24, 4.4.19, 5.0.13, 6.0.2)
# 详情可参考如下issue中的讨论:https://github.com/alibaba/MongoShake/issues/923
full_sync.reader.split_max_chunk_size = 1024

# drop the same name of collection in dest mongodb in full synchronization
# 同步时如果目的库存在,是否先删除目的库再进行同步,true表示先删除再同步,false表示不删除。
full_sync.collection_exist_drop = false

# create index option.
# none: do not create indexes.
# foreground: create indexes when data sync finish in full sync stage.
# background: create indexes when starting.
# 全量期间数据同步完毕后,是否需要创建索引。none表示不创建,foreground表示创建前台索引,background表示创建后台索引。
# 从4.2开始就是强制后台建,background option已经失去了其原本的语义,会被忽略。而且4.4+开始用两阶段索引创建流程,所有承载数据的节点都会
# 一起开始后台建。详情可参考:
# https://www.mongodb.com/docs/manual/core/index-creation/#comparison-to-foreground-and-background-builds
full_sync.create_index = background

# convert insert to update when duplicate key found
# 如果_id存在在目的库,是否将insert语句修改为update语句。
full_sync.executor.insert_on_dup_update = true
# filter orphan document for source type is sharding.
# 源端是sharding,是否需要过滤orphan文档
full_sync.executor.filter.orphan_document = false
# enable majority write in full sync.
# the performance will degrade if enable.
# 全量阶段写入端是否启用majority write
full_sync.executor.majority_enable = false


# do not make destination collection sharding if source collection is sharding
# 不要在目标端执行shardCollection的操作,仅在从分片到副本集的同步场景才需要设置为true
full_sync.do_not_shard_destination = false

# --------------------------- incremental sync configuration ---------------------------
# fetch method:
# oplog: fetch oplog from source mongodb (default)
# change_stream: use change to receive change event from source mongodb, support MongoDB >= 4.0.
# we recommend to use change_stream if possible.
incr_sync.mongo_fetch_method = change_stream
#incr_sync.mongo_fetch_method = oplog

# After the document is updated, the fields that only need to be updated are set to false, 
# and the contents of all documents are set to true
# 更新文档后,只需要更新的字段则设为false,需要全部文档内容则设为true
# 只在mongo_fetch_method = change_stream 模式下生效,且性能有所下降
incr_sync.change_stream.watch_full_document = true

# global id. used in active-active replication.
# this parameter is not supported on current open-source version.
# gid用于双活防止环形复制,目前只用于阿里云云上MongoDB,如果是阿里云云上实例互相同步
# 希望开启gid,请联系阿里云售后,sharding的有多个gid请以分号(;)分隔。
incr_sync.oplog.gids =

# distribute data to different worker by hash key to run in parallel.
# [auto]                decide by if there has unique index in collections.
#                               use `collection` if has unique index otherwise use `id`.
# [id]                  shard by ObjectId. handle oplogs in sequence by unique _id
# [collection]  shard by ns. handle oplogs in sequence by unique ns
# hash的方式,id表示按文档hash,collection表示按表hash,auto表示自动选择hash类型。
# 如果没有索引建议选择id达到非常高的同步性能,反之请选择collection。
incr_sync.shard_key = collection
# if shard_key is collection, and users want to improve performance when some collections
# do not have unique key.
# 对于按collection哈希,如果某些表不具有唯一索引,则可以设置按_id哈希以提高并发度。
# 用户需要确认该表不会创建唯一索引,一旦检测发现存在唯一索引,则会立刻crash退出。
# 例如,db1.collection1;db2.collection2,不支持仅指定db
incr_sync.shard_by_object_id_whitelist = 

# oplog transmit worker concurrent
# if the source is sharding, worker number must equal to shard numbers.
# 内部发送(写目的DB)的worker数目,如果机器性能足够,可以提高worker个数。
incr_sync.worker = 32

# how many writing threads will be used in one worker.
# 对于目的端是kafka等非direct tunnel,启用多少个序列化线程,必须为"incr_sync.worker"的倍数。
# 默认为"incr_sync.worker"的值。
incr_sync.tunnel.write_thread = 32

# set the sync delay just like mongodb secondary slaveDelay parameter. unit second.
# 设置目的端的延迟,比如延迟源端20分钟,类似MongoDB本身主从同步slaveDelay参数,单位:秒
# 0表示不启用
incr_sync.target_delay = 0

# memory queue configuration, plz visit FAQ document to see more details.
# do not modify these variables if the performance and resource usage can
# meet your needs.
# 内部队列的配置参数,如果目前性能足够不建议修改,详细信息参考FAQ。
# batch_queue_size:每个worker线程的队列长度,worker线程从此队列取任务
# batching_max_size:一次分发给worker的任务最多包含多少个文档
# buffer_capacity:PendingQueue队列中一个buffer至少包含的文档个数,进行序列化
# buffer_size_threshold_in_kb:PendingQueue队列中一个buffer最多包含多少个字节
# fetch_batch_size:源端拉取batch的最大条数
incr_sync.worker.batch_queue_size = 64
incr_sync.adaptive.batching_max_size = 1024
incr_sync.fetcher.buffer_capacity = 64
incr_sync.fetcher.buffer_size_threshold_in_kb = 512
incr_sync.reader.fetch_batch_size = 8192

# --- direct tunnel only begin ---
# if tunnel type is direct, all the below variable should be set
# 下列参数仅用于tunnel为direct的情况。

# oplog changes to Insert while Update found non-exist (_id or unique-index)
# 如果_id不存在在目的库,是否将update语句修改为insert语句。
incr_sync.executor.upsert = false
#incr_sync.executor.upsert = true
# oplog changes to Update while Insert found duplicated key (_id or unique-index)
# 如果_id存在在目的库,是否将insert语句修改为update语句。
incr_sync.executor.insert_on_dup_update = false
#incr_sync.executor.insert_on_dup_update = true
# options:db,none
# db. write duplicated logs to mongoshake_conflict
# 如果写入存在冲突,记录冲突的文档。选项:db, none
# db:冲突写到目的DB的mongoshake_conflict库中
incr_sync.conflict_write_to = db

# enable majority write in incremental sync.
# the performance will degrade if enable.
# 增量阶段写入端是否启用majority write
incr_sync.executor.majority_enable = false
# whether executor should bypass document validation or not.
# 增量阶段写入是否临时绕开文档验证机制
incr_sync.executor.bypass_document_validation = true
#incr_sync.executor.bypass_document_validation = false


#incr_sync.executor.dup_key_strategy = skip

# --- direct tunnel only end ---

# 特殊字段,标识源端类型,默认为空。阿里云MongoDB serverless集群请配置aliyun_serverless
special.source.db.flag =

# 特殊字段,当源库和目标库都是分片集群时,且源表和目标表片键不一致,是否跳过检验
#例如,db1.collection1;db2.collection2
skip.nsshardkey.verify =

Activity

  1. harzw commented on Sep 8, 2026

    @harzw
    Author

    change_stream 模式下

    $ ./mongoshake-stat_py3 --port=9100
    |---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|
    |             replset |        logs_get/sec |       logs_repl/sec |    logs_success/sec |             tps/sec |            lsn.time |        lsn_ack.time |       lsn_ckpt.time |            now.time |        log_size_avg |        log_size_max |      fetcher_status |
    |---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|
    |             ciotrep |                none |                none |                none |                none | 1970-01-01 08:00:00 | 1970-01-01 08:00:00 | 2026-09-08 14:00:12 | 2026-09-08 14:02:35 |              4.59KB |              1.10MB |              normal |
    |---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|
    |             ciotrep |                 292 |                   0 |                   0 |                   0 | 1970-01-01 08:00:00 | 1970-01-01 08:00:00 | 2026-09-08 14:00:12 | 2026-09-08 14:02:36 |             714.00B |              1.10MB |              normal |
    |---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|
    |             ciotrep |                 283 |                   0 |                   0 |                   0 | 1970-01-01 08:00:00 | 1970-01-01 08:00:00 | 2026-09-08 14:00:12 | 2026-09-08 14:02:37 |              8.65KB |              1.10MB |              normal |
    |---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|
    |             ciotrep |                 282 |                   0 |                   0 |                   0 | 1970-01-01 08:00:00 | 1970-01-01 08:00:00 | 2026-09-08 14:00:12 | 2026-09-08 14:02:38 |              1.29KB |              1.10MB |              normal |
    |---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|
    |             ciotrep |                 249 |                   0 |                   0 |                   0 | 1970-01-01 08:00:00 | 1970-01-01 08:00:00 | 2026-09-08 14:00:12 | 2026-09-08 14:02:39 |              1.13KB |              1.10MB |              normal |
    |---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|
    |             ciotrep |                 288 |                   0 |                   0 |                   0 | 1970-01-01 08:00:00 | 1970-01-01 08:00:00 | 2026-09-08 14:00:12 | 2026-09-08 14:02:40 |              4.13KB |              1.10MB |              normal |
    |---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|---------------------|
    
  2. changed the title [-]增量抽取数据,使用change_stream模式只有插入的时序表,一直是 write_success=0 。如果incr_sync.mongo_fetch_method转成用oplog来做增量抽取又会报错误。[/-] [+]增量抽取数据失败,使用change_stream模式增量抽取只有插入操作的时序表,一直是 write_success=0 。如果incr_sync.mongo_fetch_method转成用oplog来做增量抽取又会报错误。[/+] on Sep 9, 2026
  3. zhongli-james commented on Oct 13, 2026

    @zhongli-james
    Collaborator

    两个独立的问题,分开说。

    1. change_stream 模式同步不了时序集合,这是 MongoDB 的限制

    官方文档明确写了:

    Time series collections do not support change streams because time series collections use an optimized storage format instead of tracking changes at the document level.

    https://www.mongodb.com/docs/manual/changestreams/

    Time series collections do not support change streams.

    https://www.mongodb.com/docs/manual/core/timeseries/timeseries-limitations/

    实测确认(对照实验:同一时间窗内同时写普通集合和时序集合,7.0.37 / 7.0.43 / 8.0.4 / 8.2.12 表现完全一致):

    取数方式 时序 bucket 写入是否可见
    local.oplog.rs 可见(1 条 insert + 9 条 update 全在)
    cluster-wide $changeStream 0 条(同窗内普通集合的 3 条 insert 正常返回)
    cluster-wide + showExpandedEvents 只多出一个 create DDL,bucket 仍 0 条
    直接 watch 时序逻辑集合 服务端直接报错 Cannot run aggregation on timeseries with namespace ...

    这解释了你看到的现象:get=754333 / filter=754314 是集群里其他 namespace 的事件被正常过滤掉了,而白名单里的时序写入根本没有进入 change stream,所以 write_success=0;checkpoint 照常推进是因为 change stream 本身仍在正常返回 resume token。

    结论:时序集合的增量同步必须用 incr_sync.mongo_fetch_method = oplog。

    MongoShake 这边真正的问题是没有把这件事说出来——静默推进 checkpoint,只在指标上表现为 write_success=0,用户很难自查。已修复(c44ab30):增量启动时会显式告警,列出受影响的集合并给出改用 oplog 模式的解法。

    2. oplog 模式的报错是版本相关的服务端限制

    bypassDocumentValidation is not permitted on timeseries collections 只在部分 7.0.x 小版本上存在。用真实 bucket 文档(保留原 _id,因为 bucket 的 _id 内嵌开桶时间戳,必须与 control.min 一致)实测:

    服务端 不传该字段 false true
    7.0.37 成功 成功 成功(bypass 生效)
    7.0.43 成功 成功 code=2 bypassDocumentValidation is not permitted on timeseries collections
    8.0.4 成功 成功 成功
    8.2.12 成功 成功 成功

    你的目标端是阿里云 MongoDB 小版本 8.0.18,对应内核是 MongoDB 7.0.x(baseline 在 7.0.39 附近),正落在有这个限制的区间里。

    立即可用的绕过方案:把 incr_sync.executor.bypass_document_validation 改为 false。已在 7.0.43 上用真实 bucket 验证:false 与完全不传该字段都能正常插入(n=1,目标端逻辑视图可查到全部 5 条测量值),只有显式 true 会被拒。

    已修复(9a3c06c):写入 system.buckets.* 时不再下发 bypassDocumentValidation: true,这样该配置项对时序集合不再产生副作用,不需要用户为了同步时序集合而放弃它。

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

Labels

bugSomething isn't working

Type

Projects

No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions