使用开源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 => ["http://<CSS_IP>:9200"]
       user => "<Username>"
       password => "<Password>"
       index => "kibana_sample_data_logs"
       query => '{"query":{"range":{"@timestamp":{"gte":"now-5m","lte":"now/m"}}}}'
       docinfo => true
       size => 5000
     }
   }
   
   filter {
     mutate {
       remove_field => ["@version"]
     }
   }
   
   output {
     file {
       path => "./test_inc-%{+YYYY-MM-dd}.json"
     }
   }
   

   表1 关键配置项说明

   配置项	是否必选	说明
   input	是	input插件读取Elasticsearch集群的数据。
   hosts	是	Elasticsearch集群的访问地址。
   user	否	访问Elasticsearch集群的用户名。安全集群必填，否则无法访问Elasticsearch集群。
   password	否	访问Elasticsearch集群的密码。安全集群必填，否则无法访问Elasticsearch集群。
   index	是	指定要导出的索引名称，支持通配符（如*）匹配多个索引，多个索引以英文以逗号（,）分隔。
   query	否	定义查询条件，支持Elasticsearch的DSL查询语法。 如果全量数据导出，则无需配置该参数；如果选择部分数据导出，则可以通过配置该参数实现。 设置该参数时要注意查询条件中的字段必须是索引中存在的字段，否则无法正常匹配数据。 本示例表示按时间范围查询最近5分钟的数据。“now-5m”表示当前时间往前推5分钟，“now/m”表示当前时间的整分钟。
   docinfo	否	是否包含文档元数据（如索引名、ID等）。 true：包含文档元数据。可保留文档上下文信息，便于后续数据关联。 false：不包含文档元数据。如果仅需文档内容，建议设为“false”以减少输出体积。
   slices	否	并行分片数，用于提高数据导出效率。 默认值为1，如果需提升性能可适当增加（如5~10），但需确保集群资源充足。分片数过多可能导致资源竞争，需根据集群负载调整。
   size	否	每次请求从Elasticsearch中读取的文档数量。值越大单次请求效率越高，但会增加内存消耗。 最大值由Elasticsearch集群限制，建议根据虚拟机性能和Elasticsearch集群负载调整（如1000~10000之间）
   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”）查看导出的数据，确认数据完整性。