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.

Answer 1

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).