Я хочу загрузить данные в s3 из пожарного шланга Kinesis, отформатированного как паркет. Пока я только что нашел решение, которое подразумевает создание EMR, но я ищу что-то более дешевое и быстрое, например, сохранить полученный json как паркет прямо из пожарного шланга или использовать функцию Lambda.
Большое спасибо, Хави.
Хорошие новости, эта функция была выпущена сегодня!
Amazon Kinesis Data Firehose может преобразовывать формат ваших входных данных из JSON в Apache Parquet или Apache ORC перед сохранением данных в Amazon S3. Parquet и ORC - это столбчатые форматы данных, которые экономят место и позволяют выполнять более быстрые запросы.
Чтобы включить, перейдите в поток Firehose и нажмите Edit. Вы должны увидеть раздел Преобразование формата записи, как на скриншоте ниже:
Подробнее см. В документации: https://docs.aws.amazon.com/firehose/latest/dev/record-format-conversion.html.
Разобравшись со службой поддержки AWS и сотней различных реализаций, я хотел бы объяснить, чего я достиг.
Наконец, я создал лямбда-функцию, которая обрабатывает каждый файл, созданный Kinesis Firehose, классифицирует мои события в соответствии с полезной нагрузкой и сохраняет результат в файлах Parquet в S3.
Сделать это непросто:
Прежде всего, вы должны создать виртуальную среду Python, включая все необходимые библиотеки (в моем случае Pandas, NumPy, Fastparquet и т. Д.). Поскольку полученный файл (который включает в себя все библиотеки и мою лямбда-функцию тяжелый, необходимо запустить экземпляр 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.zip и разверните lambda_function:
Запустите функцию, которая будет выполняться, когда захотите, например, каждый раз, когда новый файл создается в S3, или даже вы можете связать лямбда-функцию с Firehose. (Я не выбирал этот вариант, потому что пределы 'лямбда' ниже, чем ограничения Firehose, вы можете настроить Firehose для записи файла каждые 128 МБ или 15 минут, но если вы свяжете эту лямбда-функцию с Firehose, лямбда-функция будет выполнена каждые 3 минуты или 5 МБ, в моем случае у меня была проблема с генерацией большого количества маленьких паркетных файлов, поскольку каждый раз, когда запускается лямбда-функция, я генерирую не менее 10 файлов).
Amazon Kinesis Firehose получает записи потоковой передачи и может хранить их в Amazon S3 (или Amazon Redshift или Amazon Elasticsearch Service).
Каждая запись может иметь размер до 1000 КБ.
Однако записи добавляются вместе в текстовый файл с группировкой в зависимости от времени или размера. Традиционно записи имеют формат JSON.
Вы не сможете отправить файл паркета, потому что он не соответствует этому формату файла.
Можно запустить функцию преобразования лямбда-данных, но она также не сможет выводить файл паркета.
Фактически, учитывая природу паркетных файлов, маловероятно, что вы сможете создавать их по одной записи за раз. Я подозреваю, что поскольку они представляют собой столбчатый формат хранения, их действительно нужно создавать в пакетном режиме, а не добавлять данные для каждой записи.
Итог: Нет.
