Building Serverless Data Stream pipeline using Kinesis data streams and Firehose for Snowflake 

One of the toughest challenges data professionals face today is streaming data for real-time analytics.
A main barrier to real-time insights remains the complexity of the data itself, where companies do not have the tools and infrastructure to ingest and process structured and unstructured data.
Organizations today need a data warehouse that is able to handle all data types and scale quickly to address growth.
Here is one complete pipeline on Velocity component in Big Data where I have explained how to create Streaming pipeline from scratch
Steps:
-----------
Step 1:Create the Lambda Role
Step 2:Create the Lambda Function to read the data from API Gateway & put in Kinesis Data Stream
Step 3: Create API Gateway & make the integration with AWS Lambda created in Step 2
Step 4:Create the Kinesis Data Stream to consume data from AWS Lambda created in Step 2
Step 5:Create Lambda for processing the data before s3 dump
Step 6:Create firehose Destination s3 bucket
Step 7:Create Kinesis Firehose
Step 8:Create Snowflake Role
Lambda for Step 2:
-------------------------------
import json
import datetime
import random
import boto3
client = boto3.client('kinesis')
def lambda_handler(event, context):
TODO implement
data = json.dumps(event['body'])
client.put_record(StreamName="hellotesting", Data=data, PartitionKey="1")
print("Data Inserted")
Lambda for Firehose Transformation(Step 5):
-------------------------------------------------------------------------
import json
import boto3
import base64
output = []
def lambda_handler(event, context):
print(event)
for record in event['records']:
payload = base64.b64decode(record['data']).decode('utf-8')
print('payload:', payload)
row_w_newline = payload + "
"
print('row_w_newline type:', type(row_w_newline))
row_w_newline = base64.b64encode(row_w_newline.encode('utf-8'))
output_record = {
'recordId': record['recordId'],
'result': 'Ok',
'data': row_w_newline
}
output.append(output_record)
print('Processed {} records.'.format(len(event['records'])))
return {'records': output}
Snowflake Code:
---------------------------
--Specify the role
use role ACCOUNTADMIN;
drop database if exists s3_to_snowflake;
--Database Creation
create database if not exists s3_to_snowflake;
--Specify the active/current database for the session.
use s3_to_snowflake;
--Storage Integration Creation
create or replace storage integration s3_int
TYPE = EXTERNAL_STAGE
STORAGE_PROVIDER = S3
ENABLED = TRUE
STORAGE_AWS_ROLE_ARN = '{}'
STORAGE_ALLOWED_LOCATIONS = ('s3://{}')
COMMENT = 'Testing Snowflake getting refresh or not';
--Describe the Integration Object
DESC INTEGRATION s3_int;
--External Stage Creation
create stage mystage
url = 's3://{}'
storage_integration = s3_int;
list @mystage;
--File Format Creation
create or replace file format my_json_format
type = json;
--Table Creation
create or replace external table s3_to_snowflake.PUBLIC.Person with location = @mystage file_format ='my_json_format';
--Query the table
select parse_json(VALUE):Age as Age , trim(parse_json(VALUE):Name,'"') as Name from s3_to_snowflake.PUBLIC.Person;
Note:
----------
1)Please delete all used AWS resources if not in use else it will be creating billing!
2)As this is POC , so I gave full access for many roles creation , while moving to Production make sure to provide only that much access which is required!
3)parse_json in Snowflake Interprets an input string as a JSON document, producing a VARIANT value.
Check this playlist for more AWS Projects in Big Data domain:
• Demystifying Data Engi...
Know more about AWS Kinesis :
---------------------------------------------------
aws.amazon.com...