如何通过Logstash将Elasticsearch查询结果导出为Avro格式文件

方案1：直接使用Logstash原生组件实现（推荐） #

你对avro codec的用法存在小误区：该插件本身负责事件的Avro序列化，搭配file输出插件开启对应参数后，完全可以直接生成标准可读取的Avro格式文件，无需额外第三方插件。
如果你的Logstash还没有安装avro codec，先执行安装命令：
logstash-plugin install logstash-codec-avro
配置步骤如下：

• 第一步：提前定义你的Avro Schema文件（以.avsc为后缀，示例es_export_schema.avsc），字段需要和你ES查询返回的字段一一对应：

{
  "type": "record",
  "name": "EsExportData",
  "fields": [
    {"name": "id", "type": "string"},
    {"name": "title", "type": ["null", "string"]},
    {"name": "create_time", "type": ["null", "long"]},
    {"name": "count", "type": ["null", "int"]}
  ]
}

• 第二步：编写Logstash配置文件es_to_avro.conf：

input {
  elasticsearch {
    hosts => ["http://你的ES服务地址:9200"]
    index => "你要导出的索引名"
    query => '{"query": {"match_all": {}}}' # 替换为你自己的查询语句
    user => "ES用户名" # 无鉴权可删除
    password => "ES密码" # 无鉴权可删除
  }
}

# 可按需添加filter阶段处理字段，确保字段类型和Avro schema匹配
filter {}

output {
  file {
    path => "/你要存储Avro文件的本地路径/export_result.avro"
    codec => avro {
      schema_uri => "/你的Schema文件存放路径/es_export_schema.avsc"
      write_schema => true # 开启后会在文件头写入Schema，生成标准可直接读取的Avro数据文件
    }
  }
}

• 第三步：启动Logstash运行配置即可：
  bin/logstash -f es_to_avro.conf

方案2：二次转换实现 #

如果不想调整Logstash配置，也可以先导出JSON格式结果，再做格式转换：

• 先用Logstash file输出插件导出标准JSON行格式文件
• 用工具批量转换JSON到Avro，可选工具包括Apache Avro官方工具包avro-tools、Python的fastavro库、Java的Avro SDK等
  示例Python转换代码：

import json
from fastavro import writer, parse_schema

# 加载提前定义好的Avro schema
with open("es_export_schema.avsc", "r") as f:
    schema = parse_schema(json.load(f))

# 读取JSON行文件，写入Avro
with open("export.json", "r") as json_f, open("export.avro", "wb") as avro_f:
    writer(avro_f, schema, (json.loads(line.strip()) for line in json_f if line.strip()))

注意事项 #

• Avro schema对字段类型要求严格，若ES查询返回的字段类型和schema定义不匹配，会触发序列化报错，可在Logstash filter阶段用mutate插件提前转换字段类型
• 若需要导出大批量数据，建议在Logstash elasticsearch input中开启scroll参数，避免内存溢出

内容的提问来源于stack exchange，提问作者Sagar Kadu