使用适用于 Ruby 的 AWS 开发工具包时在 AWS Kinesis Firehose 记录之间插入换行符

Question

我有一个通过 S3 将数据发送到 Redshift 的 AWS Kinesis Firehose。 我希望在使用put_record_batch 发送的记录之间出现换行符。目前我的代码如下所示：

records = [{ id: 1, value: "foo" }, { id: 2, value: "bar" }]
Aws::Firehose::Client.new(
  region: "us-east-1"
).put_record_batch({
  delivery_stream_name: "my_firehose",
  records: records
)

最终在 S3 中的记录如下所示：

{"id":1,"value":"foo"}{"id":2,"value":"bar"}

我希望 S3 文件看起来像这样：

{"id":1,"value":"foo"}
{"id":2,"value":"bar"}

这将使在必要时手动解析文件变得更加容易（例如，如果我们需要调试为什么数据没有从 S3 传输到 Redshift）。

put_record 的解决方案很简单：将数据转换为 JSON 并添加换行符：

record = { id: 1, value: "foo" }
Aws::Firehose::Client.new(
  region: "us-east-1"
).put_record({
  delivery_stream_name: "my_firehose",
  data: record.to_json << "\n"
)

我尝试用put_record_batch做类似的事情：

records = [{ id: 1, value: "foo"}, { id: 2, value: "bar" }]
json_records = records.map { |record| record.to_json << "\n" }
Aws::Firehose::Client.new(
  region: "us-east-1"
).put_record_batch({
  delivery_stream_name: "my_firehose",
  records: json_records
)

但这导致了错误：

ArgumentError: parameter validator found 2 errors:
  - expected params[:records][0] to be a hash, got value "{\"id\":1,\"value\":\"foo\"}\n" (class: String) instead.
  - expected params[:records][1] to be a hash, got value "{\"id\":2,\"value\":\"bar\"}\n" (class: String) instead.
from /mnt/istore/apps/my_app/shared/bundle/ruby/2.7.0/gems/aws-sdk-core-3.89.1/lib/aws-sdk-core/param_validator.rb:33:in `validate!'

看来我们需要发送一个哈希。

put_record_batch 的文档说：

Kinesis Data Firehose 在将记录传送到目的地之前对其进行缓冲。为了消除目标数据 blob 的歧义，一种常见的解决方案是在数据中使用分隔符，例如换行符 (\n) 或数据中唯一的一些其他字符。这允许消费者应用程序在从目标读取数据时解析单个数据项。

我该怎么做？

我正在使用 aws-sdk-firehose gem 的 version 1.26.0。

Answer 1

我认为问题在于我在使用 put_record_batch 时忽略了 data 键。这似乎有效：

records = [{ id: 1, value: "foo"}, { id: 2, value: "bar" }]
json_records = records.map do |record|
  # Previously this line was `record.to_json << "\n"`
  { data: record.to_json << "\n" }
end
Aws::Firehose::Client.new(
  region: "us-east-1"
).put_record_batch({
  delivery_stream_name: "my_firehose",
  records: json_records
)