如何使用LogStash将数据迁移到Elasticsearch

一、背景

    首先介绍下ES常用的数据导入工具，以及华为云上ES的应用场景：

     1）常用数据导入方法：将数据导入到ES的方法较多，常用的工具与方法包括：Elasticsearch-dump，通过Snapshot，Beats以及LogStash等等；其中Elasticsearch-dump主要用于ES集群间导入导出数据，Snapshot主要用于ES集群间数据的备份与恢复，Beats与LogStash作为独立的ETL工具，其应用场景较广泛。

     2）华为云ES应用场景：华为云上的ES服务运行在独立的VPC中，通过外网无法直接访问，因此其主要使用场景是在ES集群所在VPC下申请一台ECS并绑定弹性IP地址（用于部署业务应用），然后客户通过EIP访问业务应用，业务应用程序再调用ES集群。具体如下图：

二、具体使用场景

    【conf文件组成】使用Logstash将数据导入到ES集群，其关键在于如何配置其conf文件。一般而言，Logstash的配置文件包括3个部分，分别是input，filter，以及output；其中input用于配置数据源，filter用于对数据进行ETL预处理，output用于配置目的端。

    【运行Logstash】当配置好conf文件后（假设其名称为logstash-simple.conf），接下来在logstash目录下，通过如下命令可启动数据导入：

./bin/logstash -f logstash-simple.conf；

    【具体使用场景】接下来我们结合几个具体需求来介绍如何配置conf文件：

1）如何指定导入的index与type名称？

    使用Logstash导入数据的过程中，如果没有指定index以及type，默认使用logstash-%{+YYYY.MM.dd}以及logs；如果想指定index以及type，可以在output中增加如下内容，使得index以myindex开头，type为mytype：

output {
    elasticsearch {
            hosts => "192.168.0.65:9200"
            index => "myindex-%{+YYYY.MM.dd}"
            document_type => "mytype"
           }
}

2）如何删除导入过程中产生的新字段？

    使用Logstash在进行数据导入的过程中，默认会添加两个字段@version，@timestamp；如果在应用中不想要这两这个字段，可以在filter中增加如下内容：

filter {
     mutate {
         remove_field => ["@version","@timestamp"]
         }
}

3）如何使得索引名称与当前时区一致？

    使用Logstash在进行数据导入的过程中，index名称后缀默认为@timestamp字段所对应的UTC时间，该设计的初衷是为了统一不同时区的查询（不同时区的查询经过Kibana首先会转化为UTC时间，然后Kibana再向ES发送查询请求，这样不同时区查询过去一段时间的数据时，获得的结果是一样的），但这种情况下如果要查询本地时间某天的数据，则需要遍历Day-1，Day以及Day+1 3天的index；为了兼顾统一查询以及查询效率，可以在filter中增加如下内容，这里假设源数据中存在一个时间字段timestamp，并且其格式为dd/MMM/yyyy:HH:mm:ss +0800（其他格式类似），那么通过如下命令可以将timestamp字段向前滚动8个小时，然后赋值给@timestamp，从而使得index名称的后缀为东八区的时间，同时通过使用timestamp进行统计分析，仍然可以统一查询。

filter {
    date {
        match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss +0800"]
        locale => "en"
        timezone => "UTC"
       }
}

4）如何将外网数据导入到华为云ES，并控制数据导入速率？

      如前所述，将外网数据导入到华为云ES，需要一台带有EIP的ECS作为中转，此时根据Logstash部署的位置，可分为两种场景：一种是Logstash部署在外网，一种是Logstash部署在ECS上。

    a) 当Logstash部署在外网时，首先需要申请一台与ES集群在同一VPC下的ECS，并开放其安全组的9200外网访问权限；然后为该ECS绑定一个EIP，并限制该EIP的网络带宽；接下来在ECS上做一个端口映射，具体命令如下，然后输入root用户密码完成端口转发（其中9200为EIP接入端口，192.168.0.66:9200为ES集群接入点，192.168.0.200为ECS本机的内网ip地址）：

ssh -g -L 9200:192.168.0.66:9200 -N -f root@192.168.0.200

最后配置Logstash的conf文件output部分如下，然后运行Logstash开始数据导入过程：

output {
    elasticsearch {
            hosts => "EIP:9200"
           }
}

    b) 当Logstash部署在ECS上时，首先需要确保ES集群与ECS在同一VPC下，并且已开放安全组的9200外网访问权限；然后为该ECS绑定一个EIP，并限制该EIP的网络带宽；最后按常规方式配置conf文件，并运行Logstash开始数据导入过程（不再需要在ECS上做端口映射）。

三、拓展思考

上面给大家介绍了如何使用Logstash将数据导入到ES集群，此外Logstash还可以用于ES集群内部index的变换处理，包括index合并、过滤某些字段等等。另外如果是在两个ES集群间互导数据，可通过指定document_id防止重复导入数据。