0 0
Read Time:2 Minute, 0 Second

During the last post we discussed about the several components used in the data pipeline flow. We saw how the different services merged into one flow and finally ingested data into snowflake. Me and my colleague Mehani Hakim worked on the technical aspects of this data Pipeline flow . Here we will see implementation part along with the Code we have used in the process.

Following Component has been used in this pipeline:

Source bucket: salessnow
Bucket where source json file will be uploaded

Lambda function: kinesislambda
Lambda to read JSON file from S3 bucket and trigger the Kinesis data stream

Kinesis Stream: ksstream
Stream to capture the JSON file

Kinesis Firehose: firehosestream
Receive input from Kinesis stream and store output to kinesisdest bucket

Firehose Destination Bucket : kinesisdest
Bucket to store the file by Firehose

Lambda: kinesistoglue
Lambda triggeres once file uploaded to kinesisdest bucket
Call the Glue job whcih will convert JSON to Parque

Job to convert JSON to Parquet and store file to bucket i.e. parquebucket

snowpipe: kinesispipe

Parquet Table: PARQUET_TABLE

Lambda to Call Kinesis


Firehose bucket
Glue Parquet

5.  Glue Script:

import sys

from awsglue.transforms import *

from awsglue.utils import getResolvedOptions

from pyspark.context import SparkContext

from awsglue.context import GlueContext

from awsglue.job import Job

args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()

glueContext = GlueContext(sc)

spark = glueContext.spark_session

job = Job(glueContext)

job.init(args['JOB_NAME'], args)

datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "kinesis_db", table_name = "demo_json_json", transformation_ctx = "datasource0")

datasink4 = glueContext.write_dynamic_frame.from_options(frame = datasource0, connection_type = "s3", connection_options = {"path": "s3://parquebucket"}, format = "parquet", transformation_ctx = "datasink4")


Parquet Bucket

7. Lambda : To Call Glue Service

import json

import boto3

def lambda_handler(event, context):

s3_client = boto3.client("glue");


# TODO implement

return {

'statusCode': 200,

'body': json.dumps('Hello from Lambda!')


8. Snowflake: Finally see below snowflake commands to load data into snowflake. Moreover, Schema detection feature is used to generate the Parquet structure at run time instead of creating the table manually.

Average Rating

5 Star
4 Star
3 Star
2 Star
1 Star

Leave a Reply

Your email address will not be published. Required fields are marked *