如何通过Laravel命令实现Elasticsearch到MongoDB的高效数据迁移？

一、先搞定环境依赖 #

• 通过Composer安装Elasticsearch客户端和MongoDB的Laravel适配包：

composer require elasticsearch/elasticsearch jenssegers/mongodb

• 在.env里配置连接信息：

# Elasticsearch
ELASTICSEARCH_HOST=http://localhost:9200
ELASTICSEARCH_USER=
ELASTICSEARCH_PASS=

# MongoDB
MONGODB_URI=mongodb://localhost:27017
MONGODB_DATABASE=your_target_db

• 在config/database.php的connections数组里添加MongoDB配置：

'mongodb' => [
    'driver' => 'mongodb',
    'host' => env('MONGODB_HOST', 'localhost'),
    'port' => env('MONGODB_PORT', 27017),
    'database' => env('MONGODB_DATABASE'),
    'username' => env('MONGODB_USERNAME'),
    'password' => env('MONGODB_PASSWORD'),
    'options' => [
        'database' => 'admin' // 认证数据库，按需调整
    ]
],

二、创建自定义迁移命令 #

生成Artisan命令文件：

php artisan make:command EsToMongoMigration

打开app/Console/Commands/EsToMongoMigration.php，替换成下面的高效迁移逻辑：

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Elasticsearch\Client;
use Jenssegers\Mongodb\Eloquent\Model;

class EsToMongoMigration extends Command
{
    protected $signature = 'migrate:es-to-mongo {index : Elasticsearch索引名} {collection : MongoDB集合名}';
    protected $description = '批量迁移Elasticsearch索引数据到MongoDB集合';

    protected $esClient;

    public function __construct(Client $esClient)
    {
        parent::__construct();
        $this->esClient = $esClient;
    }

    public function handle()
    {
        $esIndex = $this->argument('index');
        $mongoCollection = $this->argument('collection');

        $batchSize = 1000; // 每批处理条数，根据服务器内存调整
        $scrollId = null;
        $total = 0;

        $this->info("开始迁移ES索引[$esIndex]到Mongo集合[$mongoCollection]");

        do {
            // 初始化或滚动获取下一批数据
            $params = $scrollId ? [
                'scroll_id' => $scrollId,
                'scroll' => '30s'
            ] : [
                'index' => $esIndex,
                'scroll' => '30s',
                'size' => $batchSize,
                'body' => ['query' => ['match_all' => []]]
            ];

            $response = $this->esClient->scroll($params);
            $scrollId = $response['_scroll_id'];
            $hits = $response['hits']['hits'];

            if (empty($hits)) break;

            // 提取ES原始数据，无需json_encode/decode来回折腾
            $mongoData = array_map(function ($hit) {
                $hit['_source']['es_id'] = $hit['_id']; // 可选：把ES的ID存到Mongo里
                return $hit['_source'];
            }, $hits);

            // 批量插入Mongo，比单条插入快N倍
            Model::connection('mongodb')->collection($mongoCollection)->insertMany($mongoData);

            $total += count($hits);
            $this->line("已迁移 {$total} 条");

        } while (!empty($hits));

        // 清理ES的scroll上下文，避免资源浪费
        $this->esClient->clearScroll(['scroll_id' => $scrollId]);

        $this->info("迁移完成！共迁移 {$total} 条数据");
    }
}

三、关键优化说明 #

• 分批拉取数据：用ES的Scroll API分批获取，不会一次性把所有数据塞进内存，完美适配大数据集。
• 减少无效转换：直接用ES返回的_source数组，不需要json_encode再json_decode——MongoDB本身支持数组格式直接写入，省掉多余的性能开销。
• 批量写入Mongo：用insertMany替代循环单条插入，大幅提升写入效率。

四、执行迁移命令 #

在项目根目录运行：

php artisan migrate:es-to-mongo your_es_index_name your_mongo_collection_name

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