如何通过Kinesis Firehose向Elasticsearch多索引写入数据？

当然可以实现这个需求！我来一步步帮你解决问题：

第一步：确认Elasticsearch集群的关键配置 #

你提到的rest.action.multi.allow_explicit_index确实是核心开关，默认值就是true，但还是建议确认一下配置是否正确：

• 登录AWS控制台，进入你的Amazon Elasticsearch Service（现称OpenSearch Service）集群详情页
• 切换到配置标签页，找到高级选项区域
• 搜索rest.action.multi.allow_explicit_index，确保它的值是true（如果不是，修改后保存，集群会重启生效）

第二步：调整Kinesis Firehose的记录格式 #

虽然Firehose配置时要求填写一个默认索引，但只要开启了上面的参数，你可以在每条记录里指定_index（和_type，注意ES 7+已废弃_type，建议用_doc或省略）来覆盖默认值。

这里要注意：Firehose向ES发送数据时，遵循ES Bulk API的格式要求——每条操作需要拆成两行JSON：第一行是操作元数据（包含_index、_type等），第二行是实际的业务数据。你的现有代码问题在于没有按照Bulk格式构造记录，导致Firehose无法识别_index字段。

第三步：修正你的代码示例 #

下面是调整后的代码，我会标注关键改动：

const dataRecords = [
  { 
    // 第一条记录：指定索引和类型（ES 7+建议去掉_type，或用"_type": "_doc"）
    meta: { _index: "index1", _type: "type1" },
    data: { "value": "1" }
  },
  { 
    // 第二条记录：写入另一个索引
    meta: { _index: "index2", _type: "type1" },
    data: { "value": "2" }
  }
];

// 构造符合Firehose要求的Records数组
const params = {
  DeliveryStreamName: 'XXX',
  Records: dataRecords.map(record => {
    // 按照ES Bulk格式拼接：元数据行 + 换行 + 数据行
    const bulkRecord = `${JSON.stringify(record.meta)}\n${JSON.stringify(record.data)}`;
    return { Data: bulkRecord };
  })
};

firehose.putRecordBatch(params, (err, data) => {
  if (err) console.log(err, err.stack);
  else console.log(data);
});

关键说明： #

1. Bulk格式要求：每条Firehose记录必须是ES Bulk API的一个完整操作单元（两行JSON，用换行分隔），Firehose会自动把多个这样的记录打包成Bulk请求发送给ES。
2. ES版本兼容：如果你的ES集群是7.x及以上版本，_type已经被废弃，建议去掉这个字段，或者设置为_doc（ES默认的类型）。
3. 编码问题：Firehose会自动将字符串形式的Data进行Base64编码，所以不需要手动处理。

第四步：排查常见问题 #

如果还是无法生效，可以检查以下几点：

• IAM权限：Firehose的执行角色需要有es:ESHttpPost权限，并且权限范围要包含你要写入的所有索引（尽量用arn:aws:es:region:account-id:domain/domain-name/*来覆盖所有索引）。
• 索引存在性：确保你指定的_index已经存在于ES集群中，或者ES已经开启了自动创建索引的功能（action.auto_create_index默认是开启的）。
• 记录大小限制：Firehose单条记录最大支持1MB，批量请求总大小不超过4MB，注意不要超出限制。

内容的提问来源于stack exchange，提问作者Lior Goldemberg