使用开源Logstash批量导出CSS Elasticsearch数据

应用场景

CSS服务的Elasticsearch集群支持通过开源Logstash工具实现数据批量导出，适用于以下场景：

• 数据备份：定期全量备份索引数据，保障数据安全性。
• 数据迁移：将数据迁移到其他存储系统（如对象存储、数据库）。
• 数据清洗：导出数据后进行字段过滤、格式转换等预处理操作。
• 数据分析：将数据导出为JSON文件供离线分析使用。

如果只是导出部分数据（小于10MB），则可以直接使用Kibana的数据导出功能，操作指导请参见CSS服务中Kibana是否支持导出数据功能？

方案架构

图1 导出数据流程图

本方案基于开源Logstash实现数据导出，其核心架构分为三个组件：

• Input（读取数据）：使用logstash-input-elasticsearch插件连接CSS集群，读取指定索引数据。
• Filter（数据过滤）：可选配置，用于数据清洗（如字段过滤、格式转换）。
• Output（文件输出）：使用logstash-output-file插件将数据写入本地JSON文件。

方案优势

• 高效批量处理：支持全量导出或基于过滤条件的增量导出。
• 灵活配置：通过DSL查询语句自定义导出条件。
• 生态兼容：无缝集成Elasticsearch生态，无需额外开发。
• 低成本：基于开源工具实现，无需额外付费。

约束限制

• 导出速度受CSS集群负载和ECS虚拟机性能（CPU/内存）影响。
• 需确保CSS集群和Logstash的版本兼容，本示例基于Logstash 7.10.2版本，建议使用Elasticsearch 7.10.2。

前提条件

• 已获取CSS Elasticsearch集群的访问地址、用户名和密码。
• 已部署ECS虚拟机，且ECS和Elasticsearch之间网络互通（例如两者在同一个虚拟私有云和安全组中）。

步骤一：在客户端部署Logstash

在ECS虚拟机上部署开源Logstash。

1. 登录ECS虚拟机。

   ssh username@<ECS_IP>

2. 下载并安装开源Logstash。

   cd ~
   wget https://artifacts.elastic.co/downloads/logstash/logstash-7.10.2-linux-x86_64.tar.gz
   tar -zxvf logstash-7.10.2-linux-x86_64.tar.gz

   如果无法访问公网，请提前下载安装包并上传至ECS虚拟机，下载地址：https://artifacts.elastic.co/downloads/logstash/logstash-7.10.2-linux-x86_64.tar.gz。

3. 调整Logstash的内存配置。

   修改Logstash的JVM堆内存，默认是1GB，建议设置为虚拟机内存的一半。

   cd logstash-7.10.2
   vi config/jvm.options

   修改示例如图2所示。

   图2 修改Logstash的JVM堆内存
   
4. 优化Logstash的批量处理性能。

   修改“pipelines.yml”文件，调整批量处理大小，将“pipeline.batch.size”改为“5000”。

   vi config/pipelines.yml

   修改示例如图3所示

   图3 修改pipeline配置
   
5. 验证网络连通性。

   使用curl命令测试ECS虚拟机和Elasticsearch集群的连通性。

   curl -ik http://<CSS_IP>:9200    # 无认证访问
   curl -ik https://<CSS_IP>:9200 -u <Username>:<password>   # HTTPS安全认证访问

   能正常返回集群信息，表示连接成功。

步骤二：导出Elasticsearch数据

1. 创建Logstash配置文件。

   在Logstash的安装路径config目录下创建配置文件，例如“es2file_all.conf”。

   vi config/es2file_all.conf

2. 编写Logstash配置文件并保存。

   配置示例：

   input {
       elasticsearch {
           # 源端集群的节点访问地址，无需添加协议
           hosts => ["xxx.xxx.xxx.xxx:9200", "xxx.xxx.xxx.xxx:9200"]
           # 待迁移的源索引名称
           index => "xxx,xxx"
           query => '{"query":{"range":{"@timestamp":{"gte":"now-5m","lte":"now/m"}}}}'
           docinfo => true
           slices => 5
           size => 5000
           # 安全集群必配 (非安全集群请删除以下配置)
           # user => "xxx"           # 访问集群的用户名
           # password => "xxx"       # 用户名对应的密码
           # 源端集群启用SSL时，则需额外配置以下信息
           # ssl => true
           # ca_file => "/opt/logstash/xxx/certs"     # 用于认证源端集群的CA证书文件路径
       }
   }
   
   filter {
       # 移除一些Logstash事件添加的元数据字段
       mutate {
           remove_field => ["@version"]
     }
   }
   
   output {
       file {
           path => "./test_inc-%{+YYYY-MM-dd}.json"
     }
   }
   

   表1 关键配置项说明

   配置项	是否必选	说明
   input	是	input插件读取Elasticsearch集群的数据。
   hosts	是	源端集群的节点访问地址。支持配置多个IP地址。 取值格式：["<节点IP地址1>:<端口号>", "<节点IP地址2>:<端口号>"]
   user	否	访问源端集群的用户名。 安全模式的集群必填。
   password	否	访问源端集群的用户名对应的密码。 安全模式的集群必填。
   index	是	待全量迁移的源索引名称。 单个索引：直接输入索引名称（如“my_index”）。 多个索引：用英文逗号分隔多个索引（如“my_index1,my_index2”）。 通配符：支持通配符 *（匹配任意字符），如“myindex*”表示匹配所有以myindex开头的索引。
   query	否	定义查询条件，支持Elasticsearch的DSL查询语法。 如果全量数据导出，则无需配置该参数；如果选择部分数据导出，则可以通过配置该参数实现。 设置该参数时要注意查询条件中的字段必须是索引中存在的字段，否则无法正常匹配数据。 本示例表示按时间范围查询最近5分钟的数据。“now-5m”表示当前时间往前推5分钟，“now/m”表示当前时间的整分钟。
   docinfo	否	是否包含文档元数据（如索引名、ID等）。 true：包含文档元数据。可保留文档上下文信息，便于后续数据关联。 false：不包含文档元数据。如果仅需文档内容，建议设为“false”以减少输出体积。
   slices	否	并行分片数，将读取任务拆分为多个并行的切片，用于提高数据迁移吞吐量。 默认值：1 分片数过多可能导致资源竞争，需根据集群负载调整。在确保集群资源充足的情况下，需提升性能时，可适当增加本参数值，建议设置在2~8。
   size	否	每次请求从Elasticsearch中读取的文档数量。值越大单次请求效率越高，但会增加内存消耗。 最大值由Elasticsearch集群限制，建议根据虚拟机性能和Elasticsearch集群负载调整（如1000~10000之间）
   ssl	否	源端集群是否启用SSL。 取值范围： true：使用HTTPS协议传输数据。 false：使用HTTP协议传输数据。
   ca_file	否	用于认证源端集群的CA证书文件路径。 取值格式：<证书路径><证书名称>（如/opt/logstash/extend/certs） 需要获取目的端集群的安全证书并上传至Logstash安装路径下，此处填写安全证书的绝对路径。
   filter	否	filter插件可以基于实际需求处理数据。 本示例仅删除Logstash自动添加的字段，避免导出冗余数据。
   output	是	output插件可以指定输出文件路径，以及文件命名规则。 建议指定绝对路径以避免权限问题。

3. 执行Logstash配置文件，启动导出任务。

   cd logstash-7.10.2
   bin/logstash -f config/es2file_all.conf

4. 验证导出结果。

   进入输出文件路径，打开json文件（如“test_all-2025-06-05.json”）查看导出的数据，确认数据完整性。