使用logstash迁移ES1.x数据到ES6.x

注：如果es旧集群版本过低建议使用logstash5.2.x不然报错

[ERROR][logstash.pipeline        ] A plugin had an unrecoverable error. Will restart this plugin.
  Plugin: <LogStash::Inputs::Elasticsearch hosts=>["xxxx:9200"], index=>"xxxx", size=>1000, scroll=>"5m", docinfo=>true, id=>"0c0d97f3442bde5e47448f1ac819c918d0a89dc3-1", enable_metric=>true, codec=><LogStash::Codecs::JSON id=>"json_4252917a-5d59-4bee-8d9a-240f3a16c8ac", enable_metric=>true, charset=>"UTF-8">, query=>"{ "sort": [ "_doc" ] }", docinfo_target=>"@metadata", docinfo_fields=>["_index", "_type", "_id"], ssl=>false>
  Error: [400] {"error":"ElasticsearchIllegalArgumentException[Failed to decode scrollId]; nested: IOException[Bad Base64 input character decimal 123 in array position 0]; ","status":400}

具体解释：https://elasticsearch.cn/question/2702

1. 安装

1.1. 下载logstash

https://www.elastic.co/cn/downloads/logstash

1.1. 解压

tar -zxvf logstash-5.2.2.tar.gz -C /opt/servers/

1.2. 添加配置文件

cd /opt/servers/logstash-5.2.2/config

vim es_migration.conf

加入以下内容

input {
    elasticsearch {
        hosts => ["ip:9200"]
        index => "*"
        #index => "abc"
        #query => '{ "query": {"match_all" : {} } }'
        codec => "json"
        docinfo => true
    }
}

output {
    stdout {
      codec => rubydebug
    }
    elasticsearch {
        hosts => ["ip:9200"]#启动节点在新ES集群可用127.0.0.1
    document_type => "%{[@metadata][_type]}"
    document_id => "%{[@metadata][_id]}"
        index => "%{[@metadata][_index]}"
    }

}

2. 测试

2.1. 进入logstash目录

cd /opt/servers/logstash-5.2.2

2.2. 可直接命令启动：

./bin/logstash -e 'input { stdin { } } output { stdout {codec=>rubydebug} }'

此时可以在控制台输入内容并看到输出的内容

3. 数据迁移

3.1. 启动logstash

cd /opt/servers/logstash-5.2.2
./bin/logstash -f config/es_migration.conf

此时会看到es的数据信息，等待执行完成后到ES查看数据

3.2. 后台启动

nohup ./bin/logstash -f config/fbet_es.conf --config.reload.automatic >> /opt/server/logstash-5.2.2/logs/logstash_es.log 2>&1 &