如何将Elasticsearch中的近50万条记录迁移至MongoDB集合？

Hey there! Migrating 500k documents from Elasticsearch to MongoDB is totally manageable, and the best approach depends on your familiarity with tools, need for custom logic, and infrastructure setup. Let’s walk through the most reliable options:

1. Use Logstash (No-Code/Low-Code ETL) #

Logstash is part of the Elastic Stack, so it’s built to work seamlessly with Elasticsearch. It handles batching, pagination, and basic data transformations out of the box—perfect if you want to avoid writing custom code.

How to set it up: #

• Install Logstash (match the major version with your Elasticsearch cluster to avoid compatibility issues)
• Create a configuration file (e.g., es-to-mongo.conf) with three core sections:
  • Input: Connect to Elasticsearch, use the scroll API to fetch large datasets incrementally
  • Filter (optional): Transform fields (rename, remove, or modify data) if needed
  • Output: Send the processed data to MongoDB

Here’s a sample config:

input {
  elasticsearch {
    hosts => ["http://your-es-host:9200"]
    index => "your-source-index"
    scroll => "5m"
    size => 1000
    docinfo => true
  }
}

filter {
  # Optional: Rename Elasticsearch's _id to MongoDB's _id
  mutate {
    rename => { "[@metadata][_id]" => "_id" }
  }
}

output {
  mongodb {
    uri => "mongodb://your-mongo-host:27017"
    database => "your-target-db"
    collection => "your-target-collection"
    batch_size => 1000
  }
}

Run it with: logstash -f es-to-mongo.conf

Pros: Fast to set up, handles pagination automatically, minimal code required.
Cons: Less flexibility for complex transformations; you’ll need to learn Logstash’s DSL.

2. Write a Custom Script (Full Flexibility) #

If you need custom data transformation logic (e.g., nested field flattening, conditional filtering), writing a script in a language like Python is the way to go. You’ll use official clients for both Elasticsearch and MongoDB.

Step-by-step with Python: #

1. Install dependencies: pip install elasticsearch pymongo
2. Use Elasticsearch’s scroll API to fetch data in batches (avoids overwhelming the ES cluster)
3. Transform the data as needed
4. Use MongoDB’s bulk insertion methods to minimize network overhead

Sample script snippet:

from elasticsearch import Elasticsearch
from pymongo import MongoClient

# Initialize connections
es = Elasticsearch(["http://your-es-host:9200"])
mongo_client = MongoClient("mongodb://your-mongo-host:27017")
db = mongo_client["your-target-db"]
collection = db["your-target-collection"]

# Configure scroll parameters
scroll_time = "5m"
batch_size = 1000
index_name = "your-source-index"

# Initial scroll request
response = es.search(
    index=index_name,
    scroll=scroll_time,
    size=batch_size,
    query={"match_all": {}}
)
scroll_id = response["_scroll_id"]
hits = response["hits"]["hits"]

while hits:
    # Extract source data (ignore ES metadata unless needed)
    docs = [hit["_source"] for hit in hits]
    # Optional: Map ES _id to MongoDB _id
    for doc, hit in zip(docs, hits):
        doc["_id"] = hit["_id"]
    
    # Bulk insert to MongoDB
    collection.insert_many(docs)
    
    # Next scroll request
    response = es.scroll(scroll_id=scroll_id, scroll=scroll_time)
    scroll_id = response["_scroll_id"]
    hits = response["hits"]["hits"]

# Clean up scroll context
es.clear_scroll(scroll_id=scroll_id)
print("Migration completed!")

Pros: Full control over data transformation; easy to add error handling, retry logic, and checkpointing (for resuming if migration is interrupted).
Cons: Requires coding knowledge; you’ll need to handle edge cases like network failures or data schema mismatches.

3. Cloud-Based ETL Tools (For Managed Environments) #

If you’re using MongoDB Atlas (managed MongoDB) or Elastic Cloud, you can leverage built-in data integration tools:

• MongoDB Atlas Data Lake: Connect directly to your Elasticsearch cluster as a data source, then ingest data into an Atlas collection.
• Elastic Cloud Integrations: Use Elastic’s connectors to push data to MongoDB (though this is less common than pulling from ES).

This is great if you want a fully managed solution without maintaining infrastructure.

Critical Optimization Tips for 500k Records #

• Disable MongoDB Indexes Temporarily: Before migration, run collection.drop_indexes() (keep the _id index if needed). Rebuild indexes after migration with collection.create_indexes(...)—this drastically speeds up insertion.
• Tune Batch Sizes: Adjust size (ES) and batch_size (Mongo/Logstash) based on your server’s memory. 1000-5000 documents per batch is a safe starting point.
• Monitor Resource Usage: Keep an eye on CPU, memory, and disk I/O for both Elasticsearch and MongoDB. Avoid running the migration during peak traffic hours.
• Add Checkpointing: For custom scripts, save the last processed scroll_id or document ID to a file/database. If the migration fails, you can resume from where you left off instead of starting over.
• Validate Data Post-Migration: Compare document counts (es.count(index=index_name) vs collection.count_documents({})) and spot-check random documents to ensure all fields are correctly migrated.

内容的提问来源于stack exchange，提问作者siva prasad