使用 Transporter 在 mongoDB 和 ElasticSearch 之间同步数据

测试环境全部使用 Docker 搭建！

前提

安装好 docker 和 docker-compose 的环境，具体安装教程查看链接

下载镜像

docker pull mongo:4.0.10 # mongoDB 镜像
docker pull elastisearch:5.6.11 # elasticsearch 镜像
docker pull mobz/elasticsearch-head:5 # elasticsearch-head 镜像

创建所需的文件和目录

mkdir -p /root/softdata/elastisearch/config
cd /root/softdata/elastisearch/config
touch elastisearch.yml # elasricsearch 配置文件
cd ..
touch docker-compose.yaml # docker-compose 启动文件

编写 docker-compose 文件

version: '3'
networks: 
  mongoes:
services: 
  # mongoDB 副本集节点 1
  mongo-rs1:
    image: mongo:4.0.10
    restart: always    
    container_name:   mongo-rs1
    working_dir: /data
    command: mongod --replSet repl --directoryperdb --smallfiles
    ports:
      - 27001:27017
    volumes: 
      - /root/softdata/mongo/rs1/db/:/data/db/
      - /root/softdata/mongo/rs1/configdb/:/data/configdb/
    networks: 
      - mongoes
# mongoDB 副本集节点 2  
mongo-rs2: 
    image: mongo:4.0.10
    restart: always    
    container_name: mongo-rs2 
    working_dir: /data
    command: mongod --replSet repl --directoryperdb --smallfiles
    ports:
      - 27002:27017
    volumes: 
      - /root/softdata/mongo/rs2/db/:/data/db/
      - /root/softdata/mongo/rs2/configdb/:/data/configdb/
    networks: 
      - mongoes  
  elasticsearch:
    image: elasticsearch:5.6.11
    container_name: elasticsearch
    restart: always
    environment:     
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    ulimits:
      memlock:
        soft: -1
        hard: -1
    volumes:
      - /root/softdata/elasticsearch/data/:/usr/share/elasticsearch/data
      - /root/softdata/elasticsearch/logs/:/usr/share/elasticsearch/logs
      - /root/softdata/elasticsearch/plugins/:/usr/share/elasticsearch/plugins
      - /root/softdata/elasticsearch/elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml
    ports:
      - 27003:9200
    networks: 
      - mongoes
  es-head:
    image: mobz/elasticsearch-head:5
    container_name: es-head
    restart: always
    ports:
      - 27004:9100
    networks: 
      - mongoes

启动容器

docker-compose up -d

配置 mongoDB 副本集

docker exec -it mongo-rs1 bash
mongo
> rs.initiate( {
      _id: "repl",
     members : [
         {_id : 0, host : "mongo-rs1:27017" },
         {_id : 1, host : "mongo-rs2:27017" }
     ]
   }
)
repl:PRIMARY> rs.status() # 查看副本集状态

mongoDB写入测试数据

repl:PRIMARY> use testdb
repl:PRIMARY> db.createCollection("book")
repl:PRIMARY> db.device.ensureIndex({createTime:1})
repl:PRIMARY> var cnt = 0;
repl:PRIMARY> for(var i=0; i<100; i++){
   var dl = [];
   for(var j=0; j<500; j++){
       dl.push({
               "bookId" : "BBK-" + i + "-" + j,
               "type" : "Revision",
               "version" : "IricSoneVB"+j,
               "title" : "这是一本书哦",
               "subCount" : 10,
               "location" : "北京市石景山区八角游乐园",
               "author" : {
                     "name" : "dmego",
                     "email" : "dmeago@gmail.com",
                     "gender" : "female"
               },
               "createTime" : new Date()
           });
     }
     cnt += dl.length;
     db.book.insertMany(dl);
     print("insert ", cnt);
}

执行 db.book.stats() 查看 book集合状态。

ElasticSearch创建索引

打开网址: http://<IP>:27004 ，使用 elasticsearch-hard 连接上 elasticsearch ，然后使用 复合查询 项创建索引

PUT http://<IP>:27003/testdb
{
  "aliases": {
    "index_book": {}
  },
  "settings": {
    "index": {
      "refresh_interval": "30s",
      "number_of_shards": "5",
      "number_of_replicas": "0"
    }
  },
  "mappings": {
    "book": {
      "properties": {
        "bookId": {
          "type": "string"
        },
        "type": {
          "type": "string"
        },
        "version": {
          "type": "string"
        },
        "title": {
          "type": "string",
          "store": "true",
          "analyzer": "ik_smart"
        },
        "location": {
          "type": "string",
          "store": "true",
          "analyzer": "ik_max_word"
        },
        "author": {
          "properties": {
            "name": {
              "type": "string"
            },
            "email": {
              "type": "string"
            },
            "gender": {
              "type": "string"
            }
          }
        },
        "createTime": {
          "type": "string"
        }
      }
    }
  }
}

下载 Transporter 工具

mkdir -p /root/softdata/transporter
cd /root/softdata/transporter
wget https://github.com/compose/transporter/releases/download/v0.5.2/transporter-0.5.2-linux-amd64
mv transporter-0.5.2-linux-amd64 transporter
cp transporter /usr/bin

同步 mongoDB 数据到 elasticsearch

首先先初始化出配置文件

transporter init mongodb elasticsearch

执行上条语句之后，当前目录下会生成一个 pipeline.js 配置文件，打开编辑修改 mongoDB 和 elasticsearch 的 URL地址。

var source = mongodb({
  "uri": "mongodb://127.0.0.1:27002/testdb"  //mongodb 地址
  // "timeout": "30s",
  // "tail": false,
  // "ssl": false,
  // "cacerts": ["/path/to/cert.pem"],
  // "wc": 1,
  // "fsync": false,
  // "bulk": false,
  // "collection_filters": "{}",
  // "read_preference": "Primary"
})
var sink = elasticsearch({
  "uri": "http://127.0.0.1:27003/testdb" // elasticsearch 地址
  // "timeout": "10s", // defaults to 30s
  // "aws_access_key": "ABCDEF", // used for signing requests to AWS Elasticsearch service
  // "aws_access_secret": "ABCDEF" // used for signing requests to AWS Elasticsearch service
  // "parent_id": "elastic_parent" // defaults to "elastic_parent" parent identifier for Elasticsearch
})
t.Source("source", source, "/.*/").Save("sink", sink, "/.*/")

执行下面的命令进行数据同步

transporter run

执行完成之后，可到 es-head 上查看是否同步成功。