input {
  mongodb {
    uri  => 'mongodb://web:l8ka65##1@10.112.5.11:32222/who_aa_cen'
    placeholder_db_dir => "/tmp/logstash-mongodb/who_log"
    placeholder_db_name => "who_aa_cen"
    collection => "who_log"
    batch_size => 5000
    type => "pc-mongodb"
  }
}

output {
        elasticsearch {
                hosts => ["10.112.111.12:9400"]
                index => "pc-mongodb-%{+YYYY.MM.dd}"
        }
        stdout { codec => rubydebug }
}

1、猜想 使用 oplog.timestamp 记录同步本次，每次查询 oplog 的最新数据写入 oplog.timestamp，批量同步时，oplog.timestamp 中的时间戳不改变，所以失败后不能续传，connector 不适合批量同步数据。

参考资料

• Configuration Options
• mongo-connector实现MongoDB与elasticsearch实时同步深入详解
• Mongo-connector 原理
• 通过logstash-input-mongodb插件将mongodb数据导入ElasticSearch
• python实现MongoDB数据同步到Elasticsearch