使用 Elasticsearch 搜索 MongoDB 数据的步骤

1. 数据同步（同步 MongoDB 数据到 Elasticsearch）

首先，需要将 MongoDB 中的数据同步到 Elasticsearch。这可以通过多种方式实现，常见的方法包括使用 Logstash 或者自定义脚本来进行数据迁移。

示例使用 Logstash：

1. 安装 Logstash。
2. 创建一个配置文件 (mongo_to_es.conf)，内容如下：

conf
input {
  mongodb {
    uri => 'mongodb://localhost:27017'
    placeholder_db_dir => '/opt/logstash-mongodb/'
    placeholder_db_name => 'logstash_sqlite.db'
    collection => 'your_collection'
    batch_size => 5000
  }
}

filter {
  # 这里可以添加数据处理的filter
}

output {
  elasticsearch {
    hosts => ["localhost:9200"]
    index => "mongodb_index"
    document_type => "your_type"
  }
}

1. 运行 Logstash 配置：

bash
logstash -f mongo_to_es.conf

2. 查询设计

一旦数据同步到 Elasticsearch，就可以利用 Elasticsearch 的强大搜索功能来设计和优化查询。例如，可以利用 Elasticsearch 的全文搜索功能、聚合查询等。

示例查询：

假设我们需要在 MongoDB 的数据中搜索特定的用户信息，可以在 Elasticsearch 中这样查询：

bash
GET /mongodb_index/_search
{
  "query": {
    "match": {
      "username": "john_doe"
    }
  }
}

3. 结果处理

查询结果将以 JSON 格式返回，可以在应用程序中进一步处理这些数据以满足业务需求。

示例处理：

可以在后端服务中解析 Elasticsearch 返回的 JSON 数据，根据实际需要转换数据格式或执行其他业务逻辑。

4. 数据更新和维护

为了保持 Elasticsearch 和 MongoDB 的数据一致性，需要定期或实时同步 MongoDB 的数据更改到 Elasticsearch。这可以通过定时任务或监听 MongoDB 的变更流（Change Streams）实现。

示例使用 MongoDB Change Streams：

可以编写一个脚本或服务监听 MongoDB 的 Change Streams，一旦检测到数据变动（如增加、删除、修改），即时更新 Elasticsearch 数据。

python
import pymongo
from elasticsearch import Elasticsearch

client = pymongo.MongoClient('mongodb://localhost:27017')
db = client.your_database
collection = db.your_collection

es = Elasticsearch(['http://localhost:9200'])

change_stream = collection.watch()
for change in change_stream:
    if change['operationType'] == 'insert':
        es.index(index='mongodb_index', doc_type='your_type', id=change['documentKey']['_id'], body=change['fullDocument'])
    elif change['operationType'] == 'update':
        es.update(index='mongodb_index', doc_type='your_type', id=change['documentKey']['_id'], body={'doc': change['updateDescription']['updatedFields']})
    elif change['operationType'] == 'delete':
        es.delete(index='mongodb_index', doc_type='your_type', id=change['documentKey']['_id'])

总结

通过以上步骤，可以实现使用 Elasticsearch 来搜索和分析存储在 MongoDB 中的数据。这种方式利用了 Elasticsearch 的强大搜索和分析能力，同时保持了 MongoDB 的灵活性和强大的文档存储功能。

Elasticsearch 与 MongoDB 结合使用的一般步骤

要使用 Elasticsearch 搜索 MongoDB 中的数据，您需要执行以下步骤：

1. 数据同步 使用 MongoDB 的数据同步工具（例如MongoDB Connector for Elasticsearch或者Logstash的MongoDB插件）将数据从 MongoDB 同步到 Elasticsearch。这些工具会监听 MongoDB 的 oplog 功能来捕捉数据更改，并将这些更改实时同步到 Elasticsearch。
2. 配置同步工具 配置同步工具以确定哪些集合（collections）或文档（documents）需要被同步到 Elasticsearch。通常，这涉及到设置数据流水线（pipeline），定义字段映射（mapping），以及可能的转换和过滤。
3. 索引数据 将 MongoDB 数据索引到 Elasticsearch 中。索引是数据组织的方式，以便能够快速进行全文搜索。每个 MongoDB 文档都会变成 Elasticsearch 索引中的一个文档。
4. 查询数据 使用 Elasticsearch 的查询语言（如 Query DSL）来对索引的数据进行搜索。Elasticsearch 提供了广泛的搜索功能，包括全文搜索、复合查询、过滤器和聚合等。
5. 结果展示 将搜索结果展示给用户。这可能涉及到从 Elasticsearch 获取结果后再进行必要的后处理，以适配应用程序的展示需求。

示例案例

假设我们有一个 MongoDB 集合，存储了电子商务网站的产品信息。我们希望在 Elasticsearch 中建立一个可以进行全文搜索的商品索引。

以下是具体步骤的示例：

1. 安装 MongoDB Connector for Elasticsearch 首先，我们需要安装并配置 MongoDB Connector for Elasticsearch，这是一个官方插件，可以将 MongoDB 集合的数据实时同步到 Elasticsearch。

2. 配置同步 在 MongoDB Connector 中，我们定义了需要同步的数据库和集合，以及如何将 MongoDB 的文档结构映射到 Elasticsearch 的索引结构。例如，我们需要同步 products 集合，将产品名称、描述和价格等字段同步到 ES。

3. 监控和维护 在数据同步过程中，我们需要监控同步任务以确保数据的一致性，并对错误或中断进行相应的处理。

4. 编写搜索查询 当数据同步完毕后，我们可以使用 Elasticsearch 的查询 DSL 来编写搜索查询。例如，如果我们想搜索描述中包含“智能手机”的所有商品，可以编写如下查询：

   json
   {
     "query": {
       "match": {
         "description": "智能手机"
       }
     }
   }

5. 整合到应用程序 最后，将 Elasticsearch 的搜索功能集成到应用程序中，确保用户能够发出搜索请求，并展示搜索结果。

使用 Elasticsearch 搜索 MongoDB 的数据可以有效地提高搜索性能和用户体验，同时保持 MongoDB 的高效数据存储和管理能力。在实际操作过程中，还需要考虑数据一致性、容错和性能优化等因素。

如果您想要一个几乎实时同步和通用的解决方案，River 是一个很好的解决方案。

如果您已经在 MongoDB 中拥有数据，并且希望像“一次性”一样轻松地将其发送到 Elasticsearch，您可以在 Node.js https://github.com/itemsapi/elasticbulk中尝试我的包。

它使用 Node.js 流，因此您可以从支持流的所有内容（即 MongoDB、PostgreSQL、MySQL、JSON 文件等）导入数据

MongoDB 到 Elasticsearch 的示例：

安装包：

shell
npm install elasticbulk
npm install mongoose
npm install bluebird

创建脚本即script.js：

shell
const elasticbulk = require('elasticbulk');
const mongoose = require('mongoose');
const Promise = require('bluebird');
mongoose.connect('mongodb://localhost/your_database_name', {
  useMongoClient: true
});

mongoose.Promise = Promise;

var Page = mongoose.model('Page', new mongoose.Schema({
  title: String,
  categories: Array
}), 'your_collection_name');

// stream query 
var stream = Page.find({
}, {title: 1, _id: 0, categories: 1}).limit(1500000).skip(0).batchSize(500).stream();

elasticbulk.import(stream, {
  index: 'my_index_name',
  type: 'my_type_name',
  host: 'localhost:9200',
})
.then(function(res) {
  console.log('Importing finished');
})

发送您的数据：

shell
node script.js

它不是非常快，但它可以处理数百万条记录（感谢流）。