elasticsearch数据迁移之logstash_logstash迁移elasticseach数据

系列文章目录

第一章 es集群搭建
第二章 es集群基本操作命令
第三章 es基于search-guard插件实现加密认证
第四章 es常用插件
第五章 es数据迁移之elasticdump

---

---

前言

通过第五章内容，我们了解到了elasticdump工具只适合es数据量较小，且索引不是很多、很多的情况下适用，即大部分情况用于备份单个索引。但是在实际生产环境中，迁移整个es集群数据的次数也很多，因此elasticdump就很不适用，如果强硬使用还会导致服务器磁盘IO及CPU过高,容易产生告警，所以本篇文章再给大家推荐一个适用于生产环境的迁移工具logstash。

---

一、logstash是什么？

Logstash是具有实时流水线能力的开源的数据收集引擎。Logstash可以动态统一不同来源的数据，并将数据标准化到您选择的目标输出。它提供了大量插件，可帮助我们解析，丰富，转换和缓冲任何类型的数据

二、数据全量迁移步骤

注意: 前提是源es要与目标es网络互通

1.安装logstash

	下载合适的LogStash安装包并解压安装
    wget https://artifacts.elastic.co/downloads/logstash/logstash-7.10.0-linux-x86_64.tar.gz
	tar -zvxf logstash-7.10.0-linux-x86_64.tar.gz
1
2
3

2.修改logstash配置

    修改Logstash的堆内存使用
    	 vi config/jvm.options，修改Logstash配置文件config/jvm.options，增加-Xms2g和-Xmx2g。
    
    修改Logstash批量写入记录条数，可以加快集群数据的迁移效率
    	vi config/pipelines.yml 
    		pipeline.batch.size从125改为5000
1
2
3
4
5
6

3.创建全量迁移的logstash文件

	配置logstash
		进入到安装目录下
		cd /export/server/logstash/confing/
		创建vi es2es_all.conf文件
		input {
		  elasticsearch {
		    hosts => "http://ip:9200"  ##源es集群
		    user => "用户名"           ##认证信息
		    password => "密码"
		    index => "索引名称"  ##⽀持通配符，* 表⽰所有索引，如果索引多数据量⼤可以分开配置
		    query => '{ "sort": [ "_doc" ] }'
		    slices => 4   ##是否使⽤slice scroll加速迁移，值不超过单索引shard数
		    scroll => "5m"   ##scroll session保持时间
		    size => 1000
		    docinfo => true
		    ssl => false   ##是否使⽤ssl
		  }
		}
		filter {
  # 去掉一些Logstash自己加的字段。
		  mutate {
		    remove_field => ["@timestamp", "@version"]
		  }
		}
		output {
		  elasticsearch {
		    hosts => "http://ip:9200" ##目的 es集群
		    user => "用户名"
		    password => "密码"
		    index => "索引名称" #与源es索引保持一致即可
		    #index => "%{[@metadata][_index]}"  #根据原来的信息填写对端的信息
		    document_type => "%{[@metadata][_type]}" #目标端索引type，以下配置表示索引类型与源端保持一致
		    document_id => "%{[@metadata][_id]}" #目标端数据的id，如果不需要保留原id，可以删除以下这行，删除后性能会更好
		    ssl => false #关闭ssl
		    ssl_certificate_verification => false
		    ilm_enabled => false
 	        manage_template => false
		  }
		}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

4.执行迁移命令并检查结果

	启动Logstash全量迁移任务
		nohup bin/logstash -f config/es2es_all.conf >es_all.log 2>&1 &
	
	查看es_all.log日志是否有迁移报错,如果没有则执行以下命令,检查源索引和迁移后的目标索引大小是否一致
		curl -X GET http://ip:9200/_cat/indices?v​
1
2
3
4
5

三、数据增量迁移步骤

注意: 前提是源es要与目标es网络互通

1、创建增量迁移文件

1、创建增量迁移文件
	安装logstash及调整配置可见上述‘步骤二’
	说明:
		8.5版本Logstash的配置参数有所调整，需要去掉document_type => "%{[@metadata][_type]}"。
	按如下脚本修改Logstash配置文件后，开启Logstash定时任务即可触发增量迁移。
	vim logstash/config/es_add.conf
	input{
    elasticsearch{
        # 源端ES地址。
        hosts =>  ["http://localhost:9200"]
        # 安全集群配置登录用户名密码。
        user => "xxxxxx"
        password => "xxxxxx"
        # 需要迁移的索引列表，多个索引使用英文逗号（,）分隔。
        index => "kibana_sample_data_logs"
        # 按时间范围查询增量数据，以下配置表示查询最近5分钟的数据。
        query => '{"query":{"range":{"@timestamp":{"gte":"now-5m","lte":"now/m"}}}}'
        # 定时任务，以下配置表示每分钟执行一次。
        schedule => "* * * * *"
        scroll => "5m"
        docinfo=>true
        size => 5000
    }
}
filter {
  # 去掉一些Logstash自己加的字段.
  mutate {
    remove_field => ["@timestamp", "@version"]
  }
}
output{
    elasticsearch{
        # 目标端ES地址，可在阿里云Elasticsearch实例的基本信息页面获取。
        hosts => ["http://ip:9200"]
        # 安全集群配置登录用户名密码.
        user => "elastic"
        password => "xxxxxx"
        # 目标端索引名称，以下配置表示索引与源端保持一致。
        index => "%{[@metadata][_index]}"
        # 目标端索引type，以下配置表示索引类型与源端保持一致。
        document_type => "%{[@metadata][_type]}"
        # 目标端数据的id，如果不需要保留原id，可以删除以下这行，删除后性能会更好。
        document_id => "%{[@metadata][_id]}"
        ilm_enabled => false
        manage_template => false
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

2. 启动增量迁移并检查增量迁移是否成功

	执行以下命令
	nohup bin/logstash -f config/es_add.conf >es_add.log 2>&1 &
	
	检查是否成功
	curl -XGET http://localhost:9200/kibana_sample_data_logs/_search
	{
	  "query": {
	    "range": {
	      "@timestamp": {
	        "gte": "now-5m",
	        "lte": "now/m"
	      }
	    }
	  },
	  "sort": [
	    {
	      "@timestamp": {
	        "order": "desc"
	      }
	    }
	  ]
	}
	查看返回结果是否包含success字样，若包含则增量迁移成功。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

---

总结

本次分享的logstash迁移方法,最重要的一点就是源es和目标es网络要互相通，如果不通，这种方法会迁移失败。