This post walks you through all the steps required to build a data engineering pipeline for batch data using AWS Step Functions. The sequence of steps works like so : the ingested data arrives as a CSV file in a S3 based data lake in the landing zone, which automatically triggers a Lambda function to invoke the Step Function. I have assumed that data is being ingested daily in a .csv file with a filename_date.csv naming convention like so customers_20190821.csv. The step function, as the first step, starts a landing to raw zone file transfer operation via a Lambda Function. Then we have an AWS Glue crawler crawl the raw data into an Athena table, which is used as a source for AWS Glue based PySpark transformation script. The transformed data is written in the refined zone in the parquet format. Again an AWS Glue crawler runs to “reflect” this refined data into another Athena table. Finally, the data science team can consume this refined data available in the Athena table, using an AWS Sagemaker based Jupyter notebook instance. It is to be noted that the data science does not need to do any data pull manually, as the data engineering pipeline automatically pulls in the delta data, as per the data refresh schedule that writes new data in the landing zone.
Let’s go through the steps
How to make daily data available to Amazon SageMaker?
What is Amazon SageMaker?
Amazon SageMaker is an end-to-end machine learning (ML) platform that can be leveraged to build, train, and deploy machine learning models in AWS. Using the Amazon SageMaker Notebook module, improves the efficiency of interacting with the data without the latency of bringing it locally.
For deep dive into Amazon SageMaker, please go through the official docs.
In this blog post, I will be using a dummy customers data. The customers data consists of retailer information and units purchased.
Updating Table Definitions with AWS Glue
The data catalog feature of AWS Glue and the inbuilt integration to Amazon S3 simplifies the process of identifying data and deriving the schema definition out of the source data. Glue crawlers within Data catalog, are used to build out the metadata tables of data stored in Amazon S3.
I created a crawler named raw for the data in raw zone (s3://bucketname/data/raw/customers/). In case you are just starting out on AWS Glue crawler, I have explained how to create one from scratch in one of my earlier article. If you run this crawler, it creates customers table in specified database (raw).
Create an invocation Lambda Function
In case you are just starting out on Lambda functions, I have explained how to create one from scratch with an IAM role to access the StepFunctions, Amazon S3, Lambda and CloudWatch in my earlier article.
Add trigger to the created Lambda function named invoke-step-functions. Configure Bucket, Prefix and Suffix accordingly.
Once file is arrived at landing zone, it triggers the invoke Lambda function which extracts year, month, day from file name that comes from event. It passes year, month, day with two characters from uuid as input to the AWS StepFunctions.Please replace the following code in invoke-step-function Lambda.
import json
import uuid
import boto3
from datetime import datetime
sfn_client = boto3.client('stepfunctions')
stm_arn = 'arn:aws:states:us-west-2:XXXXXXXXXXXX:stateMachine:Datapipeline-for-SageMaker'
def lambda_handler(event, context):
# Extract bucket name and file path from event
bucket_name = event['Records'][0]['s3']['bucket']['name']
path = event['Records'][0]['s3']['object']['key']
file_name_date = path.split('/')[2]
processing_date_str = file_name_date.split('_')[1].replace('.csv', '')
processing_date = datetime.strptime(processing_date_str, '%Y%m%d')
# Extract year, month, day from date
year = processing_date.strftime('%Y')
month = processing_date.strftime('%m')
day = processing_date.strftime('%d')
uuid_temp = uuid.uuid4().hex[:2]
execution_name = '{processing_date_str}-{uuid_temp}'.format(processing_date_str=processing_date_str, uuid_temp=uuid_temp)
# Starts the execution of AWS StepFunctions
response = sfn_client.start_execution(
stateMachineArn = stm_arn,
name= str(execution_name),
input= json.dumps({"year": year, "month": month, "day": day})
)
return {"year": year, "month": month, "day": day}Create a Generic FileTransfer Lambda
Create a Lambda function named generic-file-transfer as we created earlier in this article. In the file transfer Lambda function, it transfers files from landing zone to raw zone and landing zone to archive zone based on event coming from the StepFunction.
- If step is landing-to-raw-file-transfer, the Lambda function copies files from landing to raw zone.
- If step is landing-to-archive-file-transfer, the Lambda function copies files from landing to archive zone and deletes files from landing zone.
Please replace the following code in generic-file-transfer Lambda.
import json
import boto3
s3 = boto3.resource('s3')
def lambda_handler(event, context):
# Extract Parameters from Event (invoked by StepFunctions)
step = event['step']
year = event['year']
month = event['month']
day = event['day']
bucket_name = event['bucket_name']
source_prefix = event['source_prefix']
destination_prefix = event['destination_prefix']
bucket = s3.Bucket(bucket_name)
for objects in bucket.objects.filter(Prefix = source_prefix):
file_path = objects.key
if ('.csv' in file_path) and (step == 'landing-to-raw-file-transfer'):
# Extract filename from file_path
file_name_date = file_path.split('/')[2]
file_name = file_name_date.split('_')[0]
# Add filename to the destination prefix
destination_prefix = '{destination_prefix}{file_name}/year={year}/month={month}/day={day}/'.format(destination_prefix=destination_prefix, file_name=file_name, year=year, month=month, day=day)
print(destination_prefix)
source_object = {'Bucket': bucket_name, "Key": file_path}
# Replace source prefix with destination prefix
new_path = file_path.replace(source_prefix, destination_prefix)
# Copies file
new_object = bucket.Object(new_path)
new_object.copy(source_object)
if ('.csv' in file_path) and (step == 'landing-to-archive-file-transfer'):
# Add filename to the destination prefix
destination_prefix = '{destination_prefix}{year}-{month}-{day}/'.format(destination_prefix=destination_prefix, year=year, month=month, day=day)
print(destination_prefix)
source_object = {'Bucket': bucket_name, "Key": file_path}
# Replace source prefix with destination prefix
new_path = file_path.replace(source_prefix, destination_prefix)
# Copies file
new_object = bucket.Object(new_path)
new_object.copy(source_object)
# Deletes copied file
bucket.objects.filter(Prefix = file_path).delete()
return {"year": year, "month": month, "day": day}Generic FileTransfer Lambda function setup is now complete. We need to check all files are copied successfully from one zone to another zone. If you have large files that needs to be copied, you could check out our Lightening fast distributed file transfer architecture.
Create Generic FileTransfer Status Check Lambda Function
Create a Lambda function named generic-file-transfer-status. If the step is landing to raw file transfer, the Lambda function checks if all files are copied from landing to raw zone by comparing the number of objects in landing and raw zones. If count doesn’t match it will raise an exception, and that exception is handled in AWS StepFunctions and retries after some backoff rate. If the count matches, all files are copied successfully. If the step is landing to archive file transfer, the Lambda function checks that any files are left in landing zone. Please replace the following code in generic-file-transfer-status Lambda function.
import json
import boto3
s3 = boto3.resource('s3')
def lambda_handler(event, context):
# Extract Parameters from Event (invoked by StepFunctions)
step = event['step']
year = event['year']
month = event['month']
day = event['day']
bucket_name = event['bucket_name']
source_prefix = event['source_prefix']
destination_prefix = event['destination_prefix']
bucket = s3.Bucket(bucket_name)
class LandingToRawFileTransferIncompleteException(Exception):
pass
class LandingToArchiveFileTransferIncompleteException(Exception):
pass
if (step == 'landing-to-raw-file-transfer'):
if file_transfer_status(bucket, source_prefix, destination_prefix):
print('File Transfer from Landing to Raw Completed Successfully')
else:
raise LandingToRawFileTransferIncompleteException('File Transfer from Landing to Raw not completed')
if (step == 'landing-to-archive-file-transfer'):
if is_empty(bucket, source_prefix):
print('File Transfer from Landing to Archive Completed Successfully')
else:
raise LandingToArchiveFileTransferIncompleteException('File Transfer from Landing to Archive not completed.')
return {"year": year, "month": month, "day": day}
def file_transfer_status(bucket, source_prefix, destination_prefix):
try:
# Checks number of objects at the source prefix (count of objects at source i.e., landing zone)
source_object_count = 0
for obj in bucket.objects.filter(Prefix = source_prefix):
path = obj.key
if (".csv" in path):
source_object_count = source_object_count + 1
print(source_object_count)
# Checks number of objects at the destination prefix (count of objects at destination i.e., raw zone)
destination_object_count = 0
for obj in bucket.objects.filter(Prefix = destination_prefix):
path = obj.key
if (".csv" in path):
destination_object_count = destination_object_count + 1
print(destination_object_count)
return (source_object_count == destination_object_count)
except Exception as e:
print(e)
raise e
def is_empty(bucket, prefix):
try:
# Checks if any files left in the prefix (i.e., files in landing zone)
object_count = 0
for obj in bucket.objects.filter(Prefix = prefix):
path = obj.key
if ('.csv' in path):
object_count = object_count + 1
print(object_count)
return (object_count == 0)
except Exception as e:
print(e)
raise eCreate a Generic Crawler invocation Lamda
Create a Lambda function named generic-crawler-invoke. The Lambda function invokes a crawler. The crawler name is passed as argument from AWS StepFunctions through event object. Please replace the following code in generic-crawler-invoke Lambda function.
import json
import boto3
glue_client = boto3.client('glue')
def lambda_handler(event, context):
# Extract Parameters from Event (invoked by StepFunctions)
year = event['year']
month = event['month']
day = event['day']
crawler_name = event['crawler_name']
try:
response = glue_client.start_crawler(Name = crawler_name)
except Exception as e:
print('Crawler in progress', e)
raise e
return {"year": year, "month": month, "day": day}Create a Generic Crawler Status Lambda
Create a Lambda function named generic-crawler-status. The Lambda function checks whether the crawler ran successfully or not. If crawler is in running state, the Lambda function raises an exception and the exception will be handled in the Step Function and retries after a certain backoff rate. Please replace the following code in generic-crawler-status Lambda.
import json
import boto3
glue_client = boto3.client('glue')
def lambda_handler(event, context):
class CrawlerInProgressException(Exception):
pass
# Extract Parametres from Event (invoked by StepFunctions)
year = event['year']
month = event['month']
day = event['day']
crawler_name = event['crawler_name']
response = glue_client.get_crawler_metrics(CrawlerNameList =[crawler_name])
print(response['CrawlerMetricsList'][0]['CrawlerName'])
print(response['CrawlerMetricsList'][0]['TimeLeftSeconds'])
print(response['CrawlerMetricsList'][0]['StillEstimating'])
if (response['CrawlerMetricsList'][0]['StillEstimating']):
raise CrawlerInProgressException('Crawler In Progress!')
elif (response['CrawlerMetricsList'][0]['TimeLeftSeconds'] > 0):
raise CrawlerInProgressException('Crawler In Progress!')
return {"year": year, "month": month, "day": day}Create an AWS Glue Job
AWS Glue is a fully managed ETL (extract, transform, and load) service that makes it simple and cost-effective to categorize your data, clean it, enrich it, and move it reliably between various data stores. For deep dive into AWS Glue, please go through the official docs.
Create an AWS Glue Job named raw-refined. In case you are just starting out on AWS Glue Jobs, I have explained how to create one from scratch in my earlier article. This Glue job converts file format from csv to parquet and stores in refined zone. The push down predicate is used as filter condition for reading data of only the processing date using the partitions.
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
## @params: [JOB_NAME]
# args = getResolvedOptions(sys.argv, ['JOB_NAME'])
args = getResolvedOptions(sys.argv, ['JOB_NAME', 'year', 'month', 'day'])
year = args['year']
month = args['month']
day = args['day']
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 = "raw", table_name = "customers", push_down_predicate ="((year == " + year + ") and (month == " + month + ") and (day == " + day + "))", transformation_ctx = "datasource0")
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("email_id", "string", "email_id", "string"), ("retailer_name", "string", "retailer_name", "string"), ("units_purchased", "long", "units_purchased", "long"), ("purchase_date", "string", "purchase_date", "string"), ("sale_id", "string", "sale_id", "string"), ("year", "string", "year", "string"), ("month", "string", "month", "string"), ("day", "string", "day", "string")], transformation_ctx = "applymapping1")
resolvechoice2 = ResolveChoice.apply(frame = applymapping1, choice = "make_struct", transformation_ctx = "resolvechoice2")
dropnullfields3 = DropNullFields.apply(frame = resolvechoice2, transformation_ctx = "dropnullfields3")
datasink4 = glueContext.write_dynamic_frame.from_options(frame = dropnullfields3, connection_type = "s3", connection_options = {"path": "s3://bucketname/data/refined/customers/", "partitionKeys": ["year","month","day"]}, format = "parquet", transformation_ctx = "datasink4")
job.commit()Create a Refined Crawler as we created Raw Crawler earlier in this article. Please point the crawler path to refined zone(s3://bucketname/data/refined/customers/) and database as refined. No need to create a Lambda function for refined crawler invocation and status, as we will pass crawler names from the StepFunction.
Resources required to create an the StepFunction have been created.
Creating the AWS StepFunction
StepFunction is where we create and orchestrate steps to process data according to our workflow. Create an AWS StepFunctions named Datapipeline-for-SageMaker. In case you are just starting out on AWS StepFunctions, I have explained how to create one from scratch here.
Data is being ingested into landing zone. It triggers a Lambda function which in turn invokes the execution of the StepFunction. The steps in the StepFunction are as follows:
- Transfers files from landing zone to raw zone.
- Checks all files are copied to raw zone successfully or not.
- Invokes raw Crawler which crawls data in raw zone and updates/creates definition of table in the specified database.
- Checks if the Crawler is completed successfully or not.
- Invokes Glue Job and waits for it to complete.
- Invokes refined Crawler which crawls data from refined zone in and updates/creates definition of table in the specified database.
- Checks if the Crawler is completed successfully or not.
- Transfers files from landing zone to archive zone and deletes files from landing zone.
- Checks all files are copied and deleted from landing zone successfully.
Please update the StepFunctions definition with the following code.
{
"Comment": "Datapipeline For MachineLearning in AWS Sagemaker",
"StartAt": "LandingToRawFileTransfer",
"States": {
"LandingToRawFileTransfer": {
"Comment": "Transfers files from landing zone to Raw zone.",
"Type": "Task",
"Parameters": {
"step": "landing-to-raw-file-transfer",
"bucket_name": "bucketname",
"source_prefix": "data/landing/",
"destination_prefix": "data/raw/",
"year.$": "$.year",
"month.$": "$.month",
"day.$": "$.day"
},
"Resource": "arn:aws:lambda:us-west-2:XXXXXXXXXXXX:function:generic-file-transfer",
"TimeoutSeconds": 4500,
"Catch": [
{
"ErrorEquals": [
"States.TaskFailed"
],
"Next": "LandingToRawFileTransferFailed"
},
{
"ErrorEquals": [
"States.ALL"
],
"Next": "LandingToRawFileTransferFailed"
}
],
"Next": "LandingToRawFileTransferPassed"
},
"LandingToRawFileTransferFailed": {
"Type": "Fail",
"Cause": "Landing To Raw File Transfer failed"
},
"LandingToRawFileTransferPassed": {
"Type": "Pass",
"ResultPath": "$",
"Parameters": {
"year.$": "$.year",
"month.$": "$.month",
"day.$": "$.day"
},
"Next": "LandingToRawFileTransferStatus"
},
"LandingToRawFileTransferStatus": {
"Comment": "Checks whether all files are copied from landing to raw zone successfully.",
"Type": "Task",
"Parameters": {
"step": "landing-to-raw-file-transfer",
"bucket_name": "bucketname",
"source_prefix": "data/landing/",
"destination_prefix": "data/raw/",
"year.$": "$.year",
"month.$": "$.month",
"day.$": "$.day"
},
"Resource": "arn:aws:lambda:us-west-2:XXXXXXXXXXXX:function:generic-file-transfer-status",
"Retry": [
{
"ErrorEquals": [
"LandingToRawFileTransferInCompleteException"
],
"IntervalSeconds": 30,
"BackoffRate": 2,
"MaxAttempts": 5
},
{
"ErrorEquals": [
"States.All"
],
"IntervalSeconds": 30,
"BackoffRate": 2,
"MaxAttempts": 5
}
],
"Catch": [
{
"ErrorEquals": [
"States.TaskFailed"
],
"Next": "LandingToRawFileTransferStatusFailed"
},
{
"ErrorEquals": [
"States.ALL"
],
"Next": "LandingToRawFileTransferStatusFailed"
}
],
"Next": "LandingToRawFileTransferStatusPassed"
},
"LandingToRawFileTransferStatusFailed": {
"Type": "Fail",
"Cause": "Landing To Raw File Transfer failed"
},
"LandingToRawFileTransferStatusPassed": {
"Type": "Pass",
"ResultPath": "$",
"Parameters": {
"year.$": "$.year",
"month.$": "$.month",
"day.$": "$.day"
},
"Next": "StartRawCrawler"
},
"StartRawCrawler": {
"Comment": "Crawls data from raw zone and adds table definition to the specified Database. IF table definition exists updates the definition.",
"Type": "Task",
"Parameters": {
"crawler_name": "raw",
"year.$": "$.year",
"month.$": "$.month",
"day.$": "$.day"
},
"Resource": "arn:aws:lambda:us-west-2:XXXXXXXXXXXX:function:generic-crawler-invoke",
"TimeoutSeconds": 4500,
"Catch": [
{
"ErrorEquals": [
"States.TaskFailed"
],
"Next": "StartRawCrawlerFailed"
},
{
"ErrorEquals": [
"States.ALL"
],
"Next": "StartRawCrawlerFailed"
}
],
"Next": "StartRawCrawlerPassed"
},
"StartRawCrawlerFailed": {
"Type": "Fail",
"Cause": "Crawler invocation failed"
},
"StartRawCrawlerPassed": {
"Type": "Pass",
"ResultPath": "$",
"Parameters": {
"year.$": "$.year",
"month.$": "$.month",
"day.$": "$.day"
},
"Next": "RawCrawlerStatus"
},
"RawCrawlerStatus": {
"Comment": "Checks whether crawler is successfully completed.",
"Type": "Task",
"Parameters": {
"crawler_name": "raw",
"year.$": "$.year",
"month.$": "$.month",
"day.$": "$.day"
},
"Resource": "arn:aws:lambda:us-west-2:XXXXXXXXXXXX:function:generic-crawler-status",
"Retry": [
{
"ErrorEquals": [
"CrawlerInProgressException"
],
"IntervalSeconds": 30,
"BackoffRate": 2,
"MaxAttempts": 5
},
{
"ErrorEquals": [
"States.All"
],
"IntervalSeconds": 30,
"BackoffRate": 2,
"MaxAttempts": 5
}
],
"Catch": [
{
"ErrorEquals": [
"States.TaskFailed"
],
"Next": "RawCrawlerStatusFailed"
},
{
"ErrorEquals": [
"States.ALL"
],
"Next": "RawCrawlerStatusFailed"
}
],
"Next": "RawCrawlerStatusPassed"
},
"RawCrawlerStatusFailed": {
"Type": "Fail",
"Cause": "Crawler invocation failed"
},
"RawCrawlerStatusPassed": {
"Type": "Pass",
"ResultPath": "$",
"Parameters": {
"year.$": "$.year",
"month.$": "$.month",
"day.$": "$.day"
},
"Next": "GlueJob"
},
"GlueJob": {
"Comment": "Invokes Glue job and waits for Glue job to complete.",
"Type": "Task",
"Resource": "arn:aws:states:::glue:startJobRun.sync",
"Parameters": {
"JobName": "retail-raw-refined",
"Arguments": {
"--refined_prefix": "data/refined",
"--year.$": "$.year",
"--month.$": "$.month",
"--day.$": "$.day"
}
},
"Catch": [
{
"ErrorEquals": [
"States.TaskFailed"
],
"Next": "GlueJobFailed"
},
{
"ErrorEquals": [
"States.ALL"
],
"Next": "GlueJobFailed"
}
],
"Next": "GlueJobPassed"
},
"GlueJobFailed": {
"Type": "Fail",
"Cause": "Crawler invocation failed"
},
"GlueJobPassed": {
"Type": "Pass",
"ResultPath": "$",
"Parameters": {
"year.$": "$.Arguments.--year",
"month.$": "$.Arguments.--month",
"day.$": "$.Arguments.--day"
},
"Next": "StartRefinedCrawler"
},
"StartRefinedCrawler": {
"Comment": "Crawls data from refined zone and adds table definition to the specified Database.",
"Type": "Task",
"Parameters": {
"crawler_name": "refined",
"year.$": "$.year",
"month.$": "$.month",
"day.$": "$.day"
},
"Resource": "arn:aws:lambda:us-west-2:XXXXXXXXXXXX:function:generic-crawler-invoke",
"TimeoutSeconds": 4500,
"Catch": [
{
"ErrorEquals": [
"States.TaskFailed"
],
"Next": "StartRefinedCrawlerFailed"
},
{
"ErrorEquals": [
"States.ALL"
],
"Next": "StartRefinedCrawlerFailed"
}
],
"Next": "StartRefinedCrawlerPassed"
},
"StartRefinedCrawlerFailed": {
"Type": "Fail",
"Cause": "Crawler invocation failed"
},
"StartRefinedCrawlerPassed": {
"Type": "Pass",
"ResultPath": "$",
"Parameters": {
"year.$": "$.year",
"month.$": "$.month",
"day.$": "$.day"
},
"Next": "RefinedCrawlerStatus"
},
"RefinedCrawlerStatus": {
"Comment": "Checks whether crawler is successfully completed.",
"Type": "Task",
"Parameters": {
"crawler_name": "refined",
"year.$": "$.year",
"month.$": "$.month",
"day.$": "$.day"
},
"Resource": "arn:aws:lambda:us-west-2:XXXXXXXXXXXX:function:generic-crawler-status",
"Retry": [
{
"ErrorEquals": [
"CrawlerInProgressException"
],
"IntervalSeconds": 30,
"BackoffRate": 2,
"MaxAttempts": 5
},
{
"ErrorEquals": [
"States.All"
],
"IntervalSeconds": 30,
"BackoffRate": 2,
"MaxAttempts": 5
}
],
"Catch": [
{
"ErrorEquals": [
"States.TaskFailed"
],
"Next": "RefinedCrawlerStatusFailed"
},
{
"ErrorEquals": [
"States.ALL"
],
"Next": "RefinedCrawlerStatusFailed"
}
],
"Next": "RefinedCrawlerStatusPassed"
},
"RefinedCrawlerStatusFailed": {
"Type": "Fail",
"Cause": "Crawler invocation failed"
},
"RefinedCrawlerStatusPassed": {
"Type": "Pass",
"ResultPath": "$",
"Parameters": {
"year.$": "$.year",
"month.$": "$.month",
"day.$": "$.day"
},
"Next": "LandingToArchiveFileTransfer"
},
"LandingToArchiveFileTransfer": {
"Comment": "Transfers files from landing zone to archived zone",
"Type": "Task",
"Parameters": {
"step": "landing-to-archive-file-transfer",
"bucket_name": "bucketname",
"source_prefix": "data/landing/",
"destination_prefix": "data/raw/",
"year.$": "$.year",
"month.$": "$.month",
"day.$": "$.day"
},
"Resource": "arn:aws:lambda:us-west-2:XXXXXXXXXXXX:function:generic-file-transfer",
"TimeoutSeconds": 4500,
"Catch": [
{
"ErrorEquals": [
"States.TaskFailed"
],
"Next": "LandingToArchiveFileTransferFailed"
},
{
"ErrorEquals": [
"States.ALL"
],
"Next": "LandingToArchiveFileTransferFailed"
}
],
"Next": "LandingToArchiveFileTransferPassed"
},
"LandingToArchiveFileTransferFailed": {
"Type": "Fail",
"Cause": "Crawler invocation failed"
},
"LandingToArchiveFileTransferPassed": {
"Type": "Pass",
"ResultPath": "$",
"Parameters": {
"year.$": "$.year",
"month.$": "$.month",
"day.$": "$.day"
},
"Next": "LandingToArchiveFileTransferStatus"
},
"LandingToArchiveFileTransferStatus": {
"Comment": "Checks whether all files are copied from landing to archived successfully.",
"Type": "Task",
"Parameters": {
"step": "landing-to-archive-file-transfer",
"bucket_name": "bucketname",
"source_prefix": "data/landing/",
"destination_prefix": "data/raw/",
"year.$": "$.year",
"month.$": "$.month",
"day.$": "$.day"
},
"Resource": "arn:aws:lambda:us-west-2:XXXXXXXXXXXX:function:generic-file-transfer-status",
"Retry": [
{
"ErrorEquals": [
"LandingToArchiveFileTransferInCompleteException"
],
"IntervalSeconds": 30,
"BackoffRate": 2,
"MaxAttempts": 5
},
{
"ErrorEquals": [
"States.All"
],
"IntervalSeconds": 30,
"BackoffRate": 2,
"MaxAttempts": 5
}
],
"Catch": [
{
"ErrorEquals": [
"States.TaskFailed"
],
"Next": "LandingToArchiveFileTransferStatusFailed"
},
{
"ErrorEquals": [
"States.ALL"
],
"Next": "LandingToArchiveFileTransferStatusFailed"
}
],
"Next": "LandingToArchiveFileTransferStatusPassed"
},
"LandingToArchiveFileTransferStatusFailed": {
"Type": "Fail",
"Cause": "LandingToArchiveFileTransfer invocation failed"
},
"LandingToArchiveFileTransferStatusPassed": {
"Type": "Pass",
"ResultPath": "$",
"Parameters": {
"year.$": "$.year",
"month.$": "$.month",
"day.$": "$.day"
},
"End": true
}
}
}After updating the AWS StepFunctions definition, the visual workflow looks like the following.
Now upload file in data/landing/ zone in the bucket where the trigger has been configured with the Lambda. The execution of StepFunction has started and the visual workflow looks like the following.
In RawCrawlerStatus step, if the Lambda is failing we retry till sometime and then mark the StepFunction as failed. If the StepFunction ran successfully. The visual workflow of the StepFunction looks like following.
Machine Learning workflow using Amazon SageMaker
The final step in this data pipeline is to make the processed data available in a Jupyter notebook instance of the Amazon SageMaker. Jupyter notebooks are popularly used among data scientists to do exploratory data analysis, build and train machine learning models.
Create Notebook Instance in Amazon SageMaker
Step1: In the Amazon SageMaker console choose Create notebook instance.
Step2: In the Notebook Instance settings populate the Notebook instance name, choose an instance type depends on data size, and a role for the notebook instances in Amazon SageMaker to interact with Amazon S3. The SageMaker execution role needs to have the required permission to Athena, the S3 buckets where the data resides, and KMS if encrypted.
Step3: Wait for the Notebook instances to be created and the Status to change to InService.
Step4: Choose the Open Jupyter, which will open the notebook interface in a new browser tab.
Click new to create a new notebook in Jupyter. Amazon SageMaker provides several kernels for Jupyter including support for Python 2 and 3, MXNet, TensorFlow, and PySpark. Choose Python as the kernel for this exercise as it comes with the Pandas library built in.
Step5: Within the notebook, execute the following commands to install the Athena JDBC driver. PyAthena is a Python DB API 2.0 (PEP 249) compliant client for the Amazon Athena JDBC driver.
import sys
!{sys.executable} -m pip install PyAthena
Step6: After the Athena driver is installed, you can use the JDBC connection to connect to Athena and populate the Pandas data frames. For data scientists, working with data is typically divided into multiple stages: munging and cleaning data, analyzing/ modeling it, then organizing the results of the analysis into a form suitable for plotting or tabular display. Pandas is the ideal tool for all of these tasks.
from pyathena import connect
import pandas as pd
conn = connect(s3_staging_dir='<ATHENA QUERY RESULTS LOCATION>',
region_name='REGION, for example, us-east-1')
df = pd.read_sql("SELECT * FROM <DATABASE>.<TABLENAME> limit 10;", conn)
df
As shown above, the dataframe always stays consistent with the latest incoming data because of the data engineering pipeline setup earlier in the ML workflow. This dataframe can be used for downstream ad-hoc model building purposes or for exploratory data analysis.
That’s it folks. Thanks for the read.
This story is authored by PV Subbareddy. Subbareddy is a Big Data Engineer specializing on Cloud Big Data Services and Apache Spark Ecosystem.
Comments