如何快速使用阿里云Logstash实例-检索分析服务 Elasticsearch版(ES)-阿里云帮助中心

背景信息

在开始本文操作前，请先了解以下背景信息：

前提条件

注册阿里云账号。

具体操作，请参见账号注册。
创建专有网络和虚拟交换机。

具体操作，请参见搭建IPv4专有网络。

使用限制

源Elasticsearch、Logstash和目标Elasticsearch实例在同一专有网络。如果不在同一专有网络，需要通过配置NAT网关实现与公网的连通，详细信息请参见配置NAT公网数据传输。
源Elasticsearch、Logstash和目标Elasticsearch实例版本需满足兼容性要求，详细信息请参见产品兼容性。

操作流程

步骤一：环境准备

创建源和目标Elasticsearch实例、开启目标Elasticsearch实例的自动创建索引功能、准备测试数据。
步骤二：创建阿里云Logstash实例

创建阿里云Logstash实例，等待实例状态变为正常后，才可以创建并运行管道任务。
步骤三：创建并运行管道任务

创建并配置阿里云Logstash管道任务，运行任务完成数据同步。
步骤四：查看数据同步结果

通过目标Elasticsearch实例的Kibana控制台，查看数据同步结果。

步骤一：环境准备

创建阿里云Elasticsearch实例。
1. 登录阿里云Elasticsearch控制台
2. 在左侧导航栏，单击Elasticsearch实例。
3. 在Elasticsearch实例页面，创建2个阿里云Elasticsearch实例。
  创建的2个阿里云Elasticsearch实例，分别作为Logstash的input和output，具体操作请参见创建阿里云Elasticsearch实例。本文创建的实例版本为通用商业版6.7，使用的数据迁移方案为：阿里云Elasticsearch 6.7.0 > 阿里云Logstash 6.7.0 > 阿里云Elasticsearch 6.7.0，提供的脚本仅适用于该数据迁移方案，其他方案不保证兼容。创建的阿里云Elasticsearch实例的具体配置如下。
  说明
  
  如果您使用的是其他方案，可参见产品兼容性判断是否存在兼容性问题。如果存在，可升级实例版本或新购实例。
  
  访问阿里云Elasticsearch实例的账号默认为elastic（本文以此为例），如果需要使用自建用户，要给予自建用户相应的角色和权限，详细信息请参见通过Elasticsearch X-Pack角色管理实现用户权限管控。
开启目标阿里云Elasticsearch实例的自动创建索引功能。

具体操作，请参见配置YML参数。

说明
阿里云Elasticsearch为了保证用户操作数据的安全性，默认将自动创建索引配置设置为不允许。阿里云Logstash在传输数据的时候，使用提交数据的方式创建索引，而不是Create index API的方式。所以在使用阿里云Logstash上传数据之前，需要先把集群的自动创建索引设置为允许，或提前创建好索引和Mapping。

准备测试数据。

进入源阿里云Elasticsearch实例的Kibana控制台，在Dev Tools页面的Console页签下，执行如下命令创建待同步的索引和文档。

重要

进入Kibana控制台的具体步骤，请参见登录Kibana控制台。
以下脚本以Elasticsearch 6.7版本为例，仅供测试。7.0及以上版本的示例脚本，请参见Elasticsearch快速入门。

创建名称为my_index，类型为my_type的索引。

PUT /my_index
{
    "settings" : {
      "index" : {
        "number_of_shards" : "5",
        "number_of_replicas" : "1"
      }
    },
    "mappings" : {
        "my_type" : {
            "properties" : {
              "post_date": {          
                   "type": "date"       
               },
              "tags": {
                   "type": "keyword"
               },
                "title" : {
                    "type" : "text"
                }
            }
        }
    }
}

在my_index索引中插入一个名称为1的文档。

PUT /my_index/my_type/1?pretty
{
  "title": "One", 
  "tags": ["ruby"],
  "post_date":"2009-11-15T13:00:00"
}

在my_index索引中插入一个名称为2的文档。

PUT /my_index/my_type/2?pretty
{
  "title": "Two", 
  "tags": ["ruby"],
  "post_date":"2009-11-15T14:00:00"
}

步骤二：创建阿里云Logstash实例

进入Logstash实例页面。
1. 进入阿里云Elasticsearch控制台的Logstash页面。
2. 在顶部菜单栏，选择与目标阿里云Elasticsearch实例相同的地域。
3. 在左侧导航栏，单击Logstash实例。
在Logstash实例页面，单击创建。
在购买页面的前三个配置页面，完成实例启动配置。
本文选择实例的付费模式为按量付费，版本为6.7，其余配置均保持默认。更多配置信息，请参见创建阿里云Logstash实例。
说明
- 在前期程序研发或功能测试期间，建议购买按量付费实例测试。
- 购买包年包月实例，可以享受优惠条件。
单击下一步：确认订单，预览实例配置。

配置不符合预期时，可单击图标修改。
选中服务协议，单击DataWorks独享资源（包年包月）。
提示开通成功后，单击DataWorks独享资源（包年包月）。
在顶部菜单栏，选择实例所在地域。在左侧导航栏，单击Logstash实例，进入Logstash实例页面，查看创建成功的实例。

步骤三：创建并运行管道任务

等到创建的Logstash实例状态变为正常后，您可以创建并运行管道任务同步数据。

在Logstash实例页面，单击目标实例右侧操作列下的管道管理。
在管道列表区域，单击创建管道。

输入管道ID和Config配置。

本文使用的Config配置如下。

input {
    elasticsearch {
        hosts => ["http://es-cn-0pp1f1y5g000h****.elasticsearch.aliyuncs.com:9200"]
        user => "elastic"
        password => "your_password"
        index => "*,-.monitoring*,-.security*,-.kibana*"
        docinfo => true
    }
}
filter {}
output {
    elasticsearch {
        hosts => ["http://es-cn-mp91cbxsm000c****.elasticsearch.aliyuncs.com:9200"]
        user => "elastic"
        password => "your_password"
        index => "%{[@metadata][_index]}"
        document_type => "%{[@metadata][_type]}"
        document_id => "%{[@metadata][_id]}"
    }
    file_extend {
        path => "/ssd/1/ls-cn-v0h1kzca****/logstash/logs/debug/test"
    }
}

参数	说明
hosts	阿里云Elasticsearch服务的访问地址。input中为`http://<源实例ID>.elasticsearch.aliyuncs.com:9200`；output中为`http://<目标实例ID>.elasticsearch.aliyuncs.com:9200`。
user	访问阿里云Elasticsearch服务的用户名，默认为elastic。
password	对应用户的密码。elastic用户的密码在创建实例时设定，如果忘记可进行重置，重置密码的注意事项和操作步骤请参见重置实例访问密码。
index	指定同步索引名。设置为`,-.monitoring,-.security,-.kibana`，表示同步除了`.`开头的系统索引外的所有索引。`%{[@metadata][_index]}`，表示匹配元数据中的index，即同步后索引的名称和源索引名称相同。说明系统索引一般用来存储Elasticsearch集群的监控日志，无需同步。
docinfo	设置为true，将会提取Elasticsearch文档的元信息，例如index、type和id。
document_type	指定同步后索引的类型。设置为`%{[@metadata][_type]}`，表示匹配元数据中的type，即同步后索引的类型和源索引类型相同。
document_id	指定同步后文档的ID。设置为`%{[@metadata][_id]}`，表示匹配元数据中的id，即同步后文档的ID和源文档ID相同。
file_extend	可选，用来开启调试日志功能，并通过path参数配置调试日志的输出路径。建议您配置该参数，配置后，可直接在控制台上查看输出结果。如果未配置，需要去目标端确认输出结果，再返回控制台修改，这样会耗费大量的时间和人力。详细信息，请参见使用Logstash管道配置调试功能。重要使用file_extend参数前，需要先安装logstash-output-file_extend插件。具体操作，请参见安装或卸载插件。其中的path参数默认为系统指定路径，请勿修改。您也可以单击开启配置调试获取path路径。

Config配置的结构及支持的数据类型的详细信息（不同版本支持的数据类型可能不同），请参见Structure of a Config File。

单击下一步，配置管道参数。

在配置的管道参数中，管道工作线程配置为实例的CPU核数，其他参数均为默认值。详细参数说明，请参见通过配置文件管理管道。
单击保存或者保存并部署。
- 保存：将管道信息保存在Logstash里并触发实例变更，配置不会生效。保存后，系统会返回管道管理页面。可在管道列表区域，单击操作列下的立即部署，触发实例重启，使配置生效。
- 保存并部署：保存并且部署后，会触发实例重启，使配置生效。
在创建成功提示框中，单击确认。

确认后，可在管道列表中查看创建成功的管道。等待实例变更完成，并且管道的状态显示为运行中时，表示阿里云Logstash开始执行同步任务。

步骤四：查看数据同步结果

数据同步任务配置完成并开始运行后，您可以通过目标阿里云Elasticsearch的Kibana控制台，查看数据同步结果。

登录目标阿里云Elasticsearch实例的Kibana控制台，根据页面提示进入Kibana主页。
登录Kibana控制台的具体操作，请参见登录Kibana控制台。
说明
本文以阿里云Elasticsearch 6.7.0版本为例，其他版本操作可能略有差别，请以实际界面为准。
在左侧导航栏，单击Dev Tools。

在Console中，执行如下命令查看数据同步结果。

GET /my_index/_search
{
  "query": {
    "match_all": {}
  }
}

预期结果如下。

{
  "took" : 2,
  "timed_out" : false,
  "_shards" : {
    "total" : 5,
    "successful" : 5,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : 4,
    "max_score" : 1.0,
    "hits" : [
      {
        "_index" : "my_index",
        "_type" : "doc",
        "_id" : "aKG3bHABnsveUjNW2iXy",
        "_score" : 1.0,
        "_source" : {
          "post_date" : "2009-11-15T14:00:00",
          "@version" : "1",
          "@timestamp" : "2020-02-22T11:45:52.992Z",
          "tags" : [
            "ruby"
          ],
          "title" : "Two"
        }
      },
      ...
    ]
  }
}

如果源端和目标端数据一致，表示数据同步成功。您也可以通过GET _cat/indices?v命令，查看源端和目标端相同索引的大小是否一致，来判断数据是否同步成功。

快速入门