我想将 kinesis firehose 中的数据提取到 s3 中，格式为镶木地板。到目前为止，我刚刚找到了一个暗示创建 EMR 的解决方案，但我正在寻找更便宜、更快的方法，比如直接从 firehose 将接收到的 json 存储为镶木地板或使用 Lambda 函数。
非常感谢， 哈维。
【问题讨论】：
标签： json amazon-web-services amazon-s3 parquet amazon-kinesis-firehose
Amazon Kinesis Firehose 接收流式记录并可以将它们存储在 Amazon S3（或 Amazon Redshift 或 Amazon Elasticsearch Service）中。
每条记录最大可达 1000KB。
但是，记录会一起附加到文本文件中，并根据时间或大小进行批处理。传统上，记录是 JSON 格式。
您将无法发送镶木地板文件，因为它不符合此文件格式。
可以触发 Lambda 数据转换函数，但这也不能输出 parquet 文件。
事实上，鉴于 parquet 文件的性质，您不太可能一次构建一条记录。作为一种列式存储格式，我怀疑它们确实需要批量创建，而不是为每条记录附加数据。
底线：不。
【讨论】：
在处理了 AWS 支持服务和一百种不同的实施之后，我想解释一下我所取得的成就。
最后，我创建了一个 Lambda 函数，它处理 Kinesis Firehose 生成的每个文件，根据负载对我的事件进行分类，并将结果存储在 S3 中的 Parquet 文件中。
做到这一点并不容易：
首先，您应该创建一个 Python 虚拟环境，包括所有必需的库（在我的例子中是 Pandas、NumPy、Fastparquet 等）。 由于生成的文件（包括所有库和我的 Lambda 函数很重，需要启动一个 EC2 实例，我使用了免费套餐中包含的那个）。要创建虚拟环境，请按照以下步骤操作：
正确创建 lambda_function：
import json
import boto3
import datetime as dt
import urllib
import zlib
import s3fs
from fastparquet import write
import pandas as pd
import numpy as np
import time
def _send_to_s3_parquet(df):
s3_fs = s3fs.S3FileSystem()
s3_fs_open = s3_fs.open
# FIXME add something else to the key or it will overwrite the file
key = 'mybeautifullfile.parquet.gzip'
# Include partitions! key1 and key2
write( 'ExampleS3Bucket'+ '/key1=value/key2=othervalue/' + key, df,
compression='GZIP',open_with=s3_fs_open)
def lambda_handler(event, context):
# Get the object from the event and show its content type
bucket = event['Records'][0]['s3']['bucket']['name']
key = urllib.unquote_plus(event['Records'][0]['s3']['object']['key'])
try:
s3 = boto3.client('s3')
response = s3.get_object(Bucket=bucket, Key=key)
data = response['Body'].read()
decoded = data.decode('utf-8')
lines = decoded.split('\n')
# Do anything you like with the dataframe (Here what I do is to classify them
# and write to different folders in S3 according to the values of
# the columns that I want
df = pd.DataFrame(lines)
_send_to_s3_parquet(df)
except Exception as e:
print('Error getting object {} from bucket {}.'.format(key, bucket))
raise e
将 lambda 函数复制到 lambda.zip 并部署 lambda_function：
在你喜欢的时候触发要执行的，例如，每次在 S3 中创建一个新文件时，甚至你可以将 lambda 函数关联到 Firehose。 （我没有选择这个选项，因为 'lambda' 限制低于 Firehose 限制，您可以将 Firehose 配置为每 128Mb 或 15 分钟写入一个文件，但如果您将此 lambda 函数关联到 Firehose，则会执行 lambda 函数每 3 分钟或 5MB，在我的情况下，我遇到了生成很多小拼花文件的问题，因为每次启动 lambda 函数时，我都会生成至少 10 个文件）。
【讨论】：
好消息，这个功能今天发布了！
Amazon Kinesis Data Firehose 可以转换您输入数据的格式 从 JSON 到 Apache Parquet 或 Apache ORC，然后将数据存储在 亚马逊 S3。 Parquet 和 ORC 是节省空间的柱状数据格式 并启用更快的查询
要启用，请转到您的 Firehose 流并单击 编辑。您应该会看到 记录格式转换 部分，如下图所示：
详见文档：https://docs.aws.amazon.com/firehose/latest/dev/record-format-conversion.html
【讨论】：