直接在github下载编译好的二进制文件，解压即可。

> ./monstache -v       
6.7.21

> ./monstache -h

# 我这里MongoDB为单机版7.0.12，没有配置账号密码
mongo-url = "mongodb://127.0.0.1:22017"
# 我这里的ES版本为7.10.1
elasticsearch-urls = ["http://192.168.31.181:9200"]

# 注意：
# direct-read-namespaces 这种是首次全量同步+增量同步，但是它存在一个问题，
# 如果我们使用direct-read-namespaces 这种在mongo里面有个表叫做tb11，也会因为匹配到了mydb.tb1导致tb11被同步到es里。 
# 如果使用change-stream-namespaces 则只会增量同步，不会进行首次的全量同步操作。
direct-read-namespaces = ["mydb.tb2", "mydb.tb1", "mydb.tb3"]
# change-stream-namespaces = ["mydb.tb2", "mydb.tb1", "mydb.tb3"]
# 如果要指定某个库的全部表都采集，可以使用 change-stream-namespaces = ["mydb"]
# 如果要全部库都采集，可以使用 change-stream-namespaces = [""]


gzip = true

# ES相关配置项
elasticsearch-user = "elastic"
elasticsearch-password = "UEvcV5HEcVtnHiVLo5a9"
# elasticsearch-max-conns = 4
# elasticsearch-healthcheck-timeout-startup = 15
# elasticsearch-healthcheck-timeout = 5
# elasticsearch-retry = true

# 启用HTTP服务器，貌似也没啥卵用
enable-http-server = true
http-server-addr="0.0.0.0:8080"

# 当MongoDB删除集合或者库的时，是否要同步的ES去（默认是true同步的，我这里给关闭掉了，具体看各自的情况考虑是否要开启）
dropped-collections = false
dropped-databases = false

# 当 resume 为 true 时，monstache 会将其已成功同步到 Elasticsearch 的 MongoDB 作的时间戳写入集合 monstache.monstache。它还会在启动时从该集合中读取该时间戳，以便重放由于 monstache 已停止而可能错过的事件。如果 monstache 是在设置了 cluster-name 选项的情况下启动的，则 resume 会自动打开。
resume = true

# 默认值为0，表示基于时间戳的更改流恢复。与 MongoDB API 4.0+ 兼容。
resume-strategy = 0

# 当 verbose 为 true 时，使用 enable debug logging （包括对 Elasticsearch 的请求跟踪）。这里便于演示设置为了true
verbose = true

# 当 index-files 为 true 时，monstache 会将存储在 GridFS 中的文件的原始内容作为附件类型索引到 Elasticsearch 中。默认情况下，index-files 为 false，这意味着 monstache 将仅索引与存储在 GridFS 中的文件相关的元数据。
# index-files = true

# 当文件高亮显示为 true 时，monstache 将启用在文件提取的文本中返回高亮显示的关键字的能力，以便对从 gridfs 在 Elasticsearch 中索引的文件进行查询。
# file-highlighting = true

# 如果此选项设置为 true，则 monstache 将在 Elasticsearch 中索引的源文档中包含 2 个自动字段。第一个是 oplog_ts ，它是直接从 MongoDB oplog 复制的事件的时间戳。第二个是 oplog_date，它是一个 Elasticsearch 日期字段，对应于同一事件的时间。
index-oplog-time = true

# 当 stats 和 index-stats 都为 true 时，monstache 将在 Elasticsearch 中写入有关其索引进度的统计信息
stats = true
index-stats = true

exit-after-direct-reads = false

# 元数据存储的库名（是在源端的MongoDB中建这个库存储） 
# config-database-name='monstache'

# 当给出 cluster-name 时，monstache 将进入高可用性模式。集群名称设置为相同值的进程将进行协调。集群中只有一个进程会同步更改。其他进程将处于暂停状态。如果同步更改的进程由于某种原因而关闭，则处于暂停状态的进程之一将接管并开始同步。请参阅 高可用性 以了解更多信息。
cluster-name = 'test_sync'

# 对于MongoDB文档的最大拆分数（默认将一个大集合拆成9个范围然后去并发查询）
# direct-read-split-max = 9

# 如果您的 MongoDB 数据包含 +Infinity、-Infinity、NaN 或无效日期等值，则需要将此选项设置为 true。Golang json 序列化器无法处理这些值，索引器将陷入无限循环。当 prune-invalid-json 设置为 true 时，Monstache 将删除这些字段，以便不会发生索引错误。
prune-invalid-json = true

# 人工定义MongoDB中表名和ES中索引名的映射关系（如果不定义的话，默认ES中的索引名就是库名.表名）。
# 个人观点：不要设置mapping映射，这样不便于一眼知道是这个索引对应的是哪个MongoDB的库表
# [[mapping]]
# namespace = "mydb.tb1"
# index = "tb1"
# pipeline=""  # 这里还支持es的pipeline的方法。等用到再去仔细研究吧。
# [[mapping]]
# namespace = "mydb.tb2"
# index = "tb2"

> ./monstache -f config.toml

INFO 2025/04/13 12:49:37 Successfully connected to Elasticsearch version 7.10.1
INFO 2025/04/13 12:49:37 Started monstache version 6.7.21
INFO 2025/04/13 12:49:37 Go version go1.23.6
INFO 2025/04/13 12:49:37 MongoDB go driver 1.17.2

INFO 2025/04/13 12:49:37 Elasticsearch go driver 7.0.31

INFO 2025/04/13 12:49:37 Successfully connected to MongoDB version 7.0.12-7
INFO 2025/04/13 12:49:37 Sending systemd READY=1
WARN 2025/04/13 12:49:37 Systemd notification not supported (i.e. NOTIFY_SOCKET is unset)
INFO 2025/04/13 12:49:37 Joined cluster test_sync

INFO 2025/04/13 12:49:37 Starting work for cluster test_sync

INFO 2025/04/13 12:49:37 Listening for events
INFO 2025/04/13 12:49:37 Watching changes on the deployment

INFO 2025/04/13 12:49:37 Resuming from timestamp {T:1744519777 I:4}

TRACE 2025/04/13 12:49:42 POST /_bulk HTTP/1.1
Host: 192.168.31.181:9200
User-Agent: elastic/7.0.31 (linux-amd64)

Content-Length: 280
Accept: application/json
Authorization: Basic ZWxhc3RpYzpVRXZjVjVIRWNWdG5IaVZMbzVhOQ==
Content-Type: application/x-ndjson
Accept-Encoding: gzip

当上游的mongo产生操作，则在monstache 里面的日志如下：

case1、mongodb上插入一条数据, monstache控制台日志如下：
TRACE 2025/04/13 12:56:25 POST /_bulk HTTP/1.1

Host: 192.168.31.181:9200
User-Agent: elastic/7.0.31 (linux-amd64)
Content-Length: 158
Accept: application/json
Authorization: Basic ZWxhc3RpYzpVRXZjVjVIRWNWdG5IaVZMbzVhOQ==
Content-Type: application/x-ndjson

Accept-Encoding: gzip


{"index":{"_index":"tb3","_id":"67fb43f7d4bf3f646e964037","version":7492657133196935169,"version_type":"external"}}

{"age":11,"job":"DBA","name":"zhang san"}


TRACE 2025/04/13 12:56:25 HTTP/1.1 200 OK
Content-Length: 253
Content-Type: application/json; charset=UTF-8

{"took":12,"errors":false,"items":[{"index":{"_index":"tb3","_type":"_doc","_id":"67fb43f7d4bf3f646e964037","_version":7492657133196935169,"result":"created","_shards":{"total":2,"successful":1,"failed":0},"_seq_no":13,"_primary_term":1,"status":201}}]}




case2、mongodb上删除一条数据, monstache控制台日志如下：

TRACE 2025/04/13 12:57:35 POST /_bulk HTTP/1.1
Host: 192.168.31.181:9200
User-Agent: elastic/7.0.31 (linux-amd64)
Content-Length: 117
Accept: application/json
Authorization: Basic ZWxhc3RpYzpVRXZjVjVIRWNWdG5IaVZMbzVhOQ==
Content-Type: application/x-ndjson
Accept-Encoding: gzip

{"delete":{"_index":"tb3","_id":"67fb43f7d4bf3f646e964037","version":7492657438139613187,"version_type":"external"}}

TRACE 2025/04/13 12:57:35 HTTP/1.1 200 OK
Content-Length: 254
Content-Type: application/json; charset=UTF-8

{"took":16,"errors":false,"items":[{"delete":{"_index":"tb3","_type":"_doc","_id":"67fb43f7d4bf3f646e964037","_version":7492657438139613187,"result":"deleted","_shards":{"total":2,"successful":1,"failed":0},"_seq_no":14,"_primary_term":1,"status":200}}]}

在执行monstache刷新数据的时候，调整下ES端相关索引的刷新时间，降低对ES的压力，例如：

https://rwynn.github.io/monstache-site/advanced/#high-availability

您可以通过启动多个具有相同 cluster-name 值的进程，在高可用性模式下运行 monstache。每个进程都将加入一个集群，该集群协同工作，以确保 monstache 进程始终同步到 Elasticsearch。

高可用性的工作原理是确保在任何给定时间 MongoDB 的 monstache.cluster 集合中有一个活动进程。只有 此集合将为集群同步。 此集合中不存在的进程将暂停。 Documents 中的 monstache.cluster 集合中已分配一个 TTL。当此集合中的文档超时时，MongoDB 将从集合中删除它，并且 monstache 集群中的另一个进程将有机会写入该集合并成为新的活动进程。

注意：下面是一个最基本的配置，建议再优化下里面的参数才可以在生产去使用。

[Unit]

Description=monstache sync service

[Service]
Type=notify
ExecStart=/usr/local/bin/monstache -f /etc/monstache/config.toml
WatchdogSec=30s
Restart=always


[Install]
WantedBy=multi-user.target

详见官方文档 https://rwynn.github.io/monstache-site/advanced/#mongodb-view-replication

使用Monstache同步mongodb数据到elasticsearch

数据库

后端

大数据

Monstache是一款用于MongoDB与Elasticsearch数据同步的工具。通过配置文件可灵活设置同步策略，支持全量与增量同步，适用于各种场景。Monstache还具备高可用性、日志记录等功能，可确保数据实时同步，提升系统性能。

Elasticsearch 

数据存储

服务器

MongoDB

云数据库采购特惠，高性能、高可靠、高安全，超值优惠等你享！


mongodb

ES特惠专场，首月1折秒杀，新客首购6折起！


4核4G3M云服务器 新用户低至38元/年！

对象存储COS新用户低至1元！


文章

问答

视频

使用Monstache同步mongodb数据到elasticsearch

使用Monstache同步mongodb数据到elasticsearch

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐