Write parquet from AWS Kinesis firehose to AWS S3

Question

I would like to ingest data into S3 from Kinesis Firehose formatted as parquet. So far I have just find a solution that implies creating an EMR, but I am looking for something cheaper and faster like store the received JSON as parquet directly from Firehose or use a Lambda function.

Thank you very much, Javi.

Dusty

Answer 1

Good news, this feature was released today!

Amazon Kinesis Data Firehose can convert the format of your input data from JSON to Apache Parquet or Apache ORC before storing the data in Amazon S3. Parquet and ORC are columnar data formats that save space and enable faster queries

To enable, go to your Firehose stream and click Edit. You should see Record format conversion section as on screenshot below:

See the documentation for details: https://docs.aws.amazon.com/firehose/latest/dev/record-format-conversion.html

Sparteine

Answer 2

After dealing with the AWS support service and a hundred of different implementations, I would like to explain what I have achieved.

Finally I have created a Lambda function that process every file generated by Kinesis Firehose, classifies my events according to the payload and stores the result in Parquet files in S3.

Doing that is not very easy:

1. First of all you should create a Python virtual env, including all the required libraries (in my case Pandas, NumPy, Fastparquet, etc). As the resulted file (that includes all the libraries and my Lambda function is heavy, it is necessary to launch an EC2 instance, I have used the one included in the free tier). To create the virtual env follow these steps:

   • Login in EC2
   • Create a folder called lambda (or any other name)
   • Sudo yum -y update
   • Sudo yum -y upgrade
   • sudo yum -y groupinstall "Development Tools"
   • sudo yum -y install blas
   • sudo yum -y install lapack
   • sudo yum -y install atlas-sse3-devel
   • sudo yum install python27-devel python27-pip gcc
   • Virtualenv env
   • source env/bin/activate
   • pip install boto3
   • pip install fastparquet
   • pip install pandas
   • pip install thriftpy
   • pip install s3fs
   • pip install (any other required library)
   • find ~/lambda/env/lib*/python2.7/site-packages/ -name "*.so" | xargs strip
   • pushd env/lib/python2.7/site-packages/
   • zip -r -9 -q ~/lambda.zip *
   • Popd
   • pushd env/lib64/python2.7/site-packages/
   • zip -r -9 -q ~/lambda.zip *
   • Popd

2. Create the lambda_function propertly:

   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
   

3. Copy the lambda function to the lambda.zip and deploy the lambda_function:

   • Go back to your EC2 instance and add the lambda function desired to the zip: zip -9 lambda.zip lambda_function.py (lambda_function.py is the file generated in the step 2)
   • Copy the generated zip file to S3, as it is very heavy to be deployed withough doing it through S3. aws s3 cp lambda.zip s3://support-bucket/lambda_packages/
   • Deploy the lambda function: aws lambda update-function-code --function-name --s3-bucket support-bucket --s3-key lambda_packages/lambda.zip

4. Trigger the to be executed when you like, e.g, each time a new file is created in S3, or even you could associate the lambda function to Firehose. (I did not choose this option because the 'lambda' limits are lower than the Firehose limits, you can configure Firehose to write a file each 128Mb or 15 minutes, but if you associate this lambda function to Firehose, the lambda function will be executed every 3 mins or 5MB, in my case I had the problem of generate a lot of little parquet files, as for each time that the lambda function is launched I generate at least 10 files).

Dusty

Do I understand correctly that this pipeline creates one parquet file per record? Parquet being a columnar storage, would then need some sort of a separate compaction job to reconcile those tiny parquet files into one larger one? – Sera

Hi @Tagar, it writes a parquet file each time that the lamba_handler is called and that can be configured, you can configure it to be launched every 15 minutes for instance, and that will create a file each 15 minutes with all the events received on this time. – Dusty

Answer 3

Amazon Kinesis Firehose receives streaming records and can store them in Amazon S3 (or Amazon Redshift or Amazon Elasticsearch Service).

Each record can be up to 1000KB.

However, records are appended together into a text file, with batching based upon time or size. Traditionally, records are JSON format.

You will be unable to send a parquet file because it will not conform to this file format.

It is possible to trigger a Lambda data transformation function, but this will not be capable of outputting a parquet file either.

In fact, given the nature of parquet files, it is unlikely that you could build them one record at a time. Being a columnar storage format, I suspect that they really need to be created in a batch rather than having data appended per-record.

Bottom line: Nope.

Intercellular

Hi @Javi, if this or any answer has solved your question please consider accepting it by clicking the check-mark. This indicates to the wider community that you've found a solution and gives some reputation to both the answerer and yourself. There is no obligation to do this. – Intercellular

@JohnRotenstein Could you have a lambda do a transformation on each buffered time/size batch from Firehose, and later concatenate the Parquet files together to a larger size every few hours or so? This lets you stream in JSON to Parquet via Firehose to get near-real time data in Athena, and still get the performance benefit of Parquet. – Herodotus

@cmclen, Parquet is a columnar file format. I don't think you could just append one row at a time -- it would defeat the purpose of using Parquet. – Intercellular

@JohnRotenstein you couldn't (until 12 days ago: cf Vlad's answer) rely on Firehose dumping the converted data for you to S3, but you could do the writing of the files yourself with S3FS or the like as bracana pointed out. You just need to return properly formatted rows for Firehose if you want them to appear as having succeeded (typically just add a processed_at timestamp and return the input rows as is). It's also possible to do it in a lambda directly if you don't rely on pandas which is too big of a library to be able to package it in a Lambda (50MB max). – Luteous