读取 Amazon Kinesis Firehose 流写入 s3 的数据

我正在将记录写入 Kinesis Firehose 流，该流最终由 Amazon Kinesis Firehose 写入 S3 文件。

我的记录对象看起来像

ItemPurchase {
    String personId,
    String itemId
}

写入 S3 的数据如下所示：

{"personId":"p-111","itemId":"i-111"}{"personId":"p-222","itemId":"i-222"}{"personId":"p-333","itemId":"i-333"}

没有逗号分隔。

Json 数组中没有起始括号

[

Json 数组中没有结束括号

]

我想读取此数据并获取 ItemPurchase 对象的列表。

List<ItemPurchase> purchases = getPurchasesFromS3(IOUtils.toString(s3ObjectContent))

读取这些数据的正确方法是什么？

令我难以置信的是，Amazon Firehose 以这种方式将 JSON 消息转储到 S3，并且不允许您设置分隔符或任何内容。

最终，我发现解决该问题的技巧是使用 JSON raw_decode 方法处理文本文件

这将允许您读取一堆串联的 JSON 记录，它们之间没有任何分隔符。

Python代码：

import json

decoder = json.JSONDecoder()

with open('giant_kinesis_s3_text_file_with_concatenated_json_blobs.txt', 'r') as content_file:

    content = content_file.read()

    content_length = len(content)
    decode_index = 0

    while decode_index < content_length:
        try:
            obj, decode_index = decoder.raw_decode(content, decode_index)
            print("File index:", decode_index)
            print(obj)
        except JSONDecodeError as e:
            print("JSONDecodeError:", e)
            # Scan forward and keep trying to decode
            decode_index += 1