Let\u2019s start!<\/p>\n\n\n\n
Our setup<\/h2>\n\n\n\n
In this example, we\u2019ll set up several temperature sensors to send temperature and diagnostic data to our pipeline and we\u2019ll perform different BI analyses to verify efficiency, and we\u2019ll use a Sagemaker model to check for anomalies.<\/p>\n\n\n\n
To keep things interesting we also want to grab historical data from two different locations: an S3 bucket and a Database residing on an EC2 instance in a different VPC from one of our ETL pipelines.<\/p>\n\n\n\n
We will use different ETL jobs to recover and extract cleaned data from row data and AWS Step Functions to orchestrate all the crawlers and jobs.<\/p>\n\n\n\n
Kinesis Data Firehose will continuously fetch sensors\u2019 data and with AWS Athena we will query information out of both aggregated and per-sensor data to show Graphical stats in Amazon Quicksight.<\/p>\n\n\n\n
Here is a simple schema illustrating the services involved and the complete flow.<\/p>\n\n\n\n Kinesis Data Firehose can be used to obtain near real-time data from sensors leveraging IoT Core SDK to connect to the actual devices. As seen in this article<\/a>, we can create a \u201cThing\u201d, thus generating a topic<\/strong>. By connecting to that topic<\/strong>, several devices can collect their metrics through Firehose by sending messages using the MQTT protocol<\/a>, and, should you need it, IoT Core can also manage device authentication<\/strong>.<\/p>\n\n\n\n To start sending sensors\u2019 data, we need to download the connection kit from the AWS IoT<\/a> page following in-page instructions.<\/p>\n\n\n\n Once downloaded, initialize a new Node.js project and install AWS-IoT-device-SDK<\/strong>. After that, it is possible to run the included start.sh<\/strong> script, making sure all the certificates, downloaded alongside the kit, are in the same directory. We can now create a local script to send data to a topic, passing the required modules and usingdevice.publish (“<topic>”, payload)<\/strong>:<\/p>\n\n\n\n The data sent is structured in JSON format with the following structure:<\/p>\n\n\n\n To create a Firehose delivery stream go to the Kinesis firehose<\/strong> service dashboard in the AWS web console, click \u201cCreate delivery stream\u201d, select a name, and then \u201cDirect PUT or other sources\u201d like in figure:<\/p>\n\n\n\n Leave \u201cData transformation\u201d and \u201cRecord format conversion\u201d as default. Choose an S3 destination as the target. Remember to also define an IoT Rule<\/strong> to send IoT messages to a Firehose delivery stream.<\/p>\n\n\n\n AWS Glue can be used to Extract and Transform data from a multitude of different data sources, thanks to the possibility of defining different types of connectors.<\/p>\n\n\n\n Database on EC2 instance<\/strong><\/p>\n\n\n\n We want to be able to generate a Glue Data Catalog from a Microsoft SQL Server DB residing on an EC2 Instance in another VPC. To do so we need to create a JDBC connection, which can be done easily by going to the AWS Glue service page and by adding a new connection, found under the \u201cData Catalog – Databases\u201d section of the sidebar menu.<\/p>\n\n\n\n Just add a name to the connection (which will be used by the related Crawler Job), the JDBC URL, following the right convention for ORACLE DBs, username and password, and the required VPC and subnet.<\/p>\n\n\n\n In order to establish a glue connection to the database, we need to create a new dedicated VPC that will be only used by Glue. The VPC is connected to the one containing the data-warehouse using VPC peering<\/a> but other options are also possible, for example, we could have used AWS Transit Gateway. Once the peering is established remember to add routes to both the Glue and the DB subnets so that they can exchange traffic and to open the DB security group to allow incoming traffic on the relevant port from the Glue security group in the new VPC.<\/p>\n\n\n\n Data on S3<\/strong><\/p>\n\n\n\n Data on S3 doesn\u2019t need a connector and can be set up directly from the AWS Glue console. Create a new crawler, selecting \u201cdata stores\u201d for the crawler source type<\/strong>; then check also \u201cCrawl all folder\u201d. After that is just a matter of setting the S3 bucket, the right IAM role and creating a new Glue Schema for this crawler. Also set \u201cRun on demand\u201d.<\/p>\n\n\n\n Glue Jobs<\/strong><\/p>\n\n\n\n Glue jobs are the steps of the ETL pipeline. They allow extracting, transforming, and saving data back to a datalake. In our example, we would like to show two different approaches: jobs managed by AWS Glue Studio<\/strong> and using custom code<\/strong>. Both jobs will be later called by AWS Step Function.<\/p>\n\n\n\n For historical data on S3, we can define Jobs from Glue Studio. For S3 select the following options in order:<\/p>\n\n\n\n In order to extract data from our EC2 instance instead, we need a custom job. To create it we need to write a script ourselves, don\u2019t worry, it\u2019s fairly easy! Here are the key points you need to know to create a Spark Job with Glue: the ETL process is composed of 6 distinct areas in the script:<\/p>\n\n\n\n Import libraries<\/strong><\/p>\n\n\n\n Basic set needed for a script:<\/p>\n\n\n\n Prepare connectors and other variables<\/strong><\/p>\n\n\n\n To be used inside the script:<\/p>\n\n\n\n Get Dynamic Frames out of a Glue Catalog obtained by a Crawler<\/strong><\/p>\n\n\n\n Use these dynamic frames to perform queries and transform data<\/p>\n\n\n\n The last line enables modifying the dynamic frame.<\/p>\n\n\n\n Apply SQL operations<\/strong><\/p>\n\n\n\n To extract distinct information<\/p>\n\n\n\n In our case, we needed to generate 3 distinct results, one for each room using a simple WHERE room_id = <value><\/strong><\/p>\n\n\n\n Apply mapping<\/strong><\/p>\n\n\n\n To generate a conversion schema<\/p>\n\n\n\n Save back to S3<\/strong><\/p>\n\n\n\n To manipulate data later on<\/p>\n\n\n\n Step function represents the core, the logic of our sample solution. Its main purpose is to manage all the ETL jobs, keep them synchronized, and manage errors. One advantage is that we can use Step Function to regulate the data being injected into the central S3 bucket which is where we save all cleaned data.<\/p>\n\n\n\n To start, this is the step function schema we used for this example:<\/p>\n\n\n\n In our example there are a couple of things we would like to share about Step Function; firstly we have 2 main crawler loops: the first one, has branches and runs 2 crawlers (one for S3 standard, and one for the EC2 database which is the custom one); the second one takes all the data retrieved from both historical data sources and live one (from Kinesis Firehose), and extract per-room datasets in order to use them with Amazon SageMaker.<\/p>\n\n\n\n As Crawlers are asynchronous we can\u2019t wait for them so we needed to create 2 waiting loops for both of the execution steps<\/p>\n\n\n\n AWS Lambda is used to call AWS Glue APIs in order to start the jobs we have configured before.<\/p>\n\n\n\n To give a hint here are some interesting parts described in the JSON file representing the state machine.<\/p>\n\n\n\n In AWS Step Function, we can launch tasks in parallel (for us, the two historical data glue jobs) using \u201cType: Parallel\u201d and \u201cBranches\u201d. Also after the key \u201cBranches\u201d, it is possible to retrieve the parallel result.<\/p>\n\n\n\n We can run a synchronous Glue job defined in the console by passing the job\u2019s name, and you can also enable the generation of a glue catalog during the process.<\/p>\n\n\n\n It is possible to catch exceptions directly in Step Function by moving to an error state using \u201cCatch\u201d:<\/p>\n\n\n\n Because we don\u2019t have a standard way to wait for the jobs to finish, we use the parallel jobs output and a StepFunctions wait cycle to check if the operation is done; for that, we use the \u201cWait\u201d key:<\/p>\n\n\n\n The rest of the flow is pretty much a repetition of these components.<\/p>\n\n\n\n The interesting fact is that we can apply some starting conditions to alter the execution of the flow, like avoiding some jobs if not needed at the moment or even run another state machine from a precise step to take our example and modularize the most complicated parts.<\/p>\n\n\n\n Athena can generate tables that can be queried using standard SQL language, not only: results of Athena queries can be imported into Amazon Quicksight to rapidly generate charts and reports, based on your data.<\/p>\n\n\n\n In our workflow, it is possible to run Athena queries on the target S3 bucket which contains both global temperature data and sensor\u2019s specific ones. Let\u2019s review quickly how to do that:<\/p>\n\n\n\n By doing these 3 steps Athena will show the result of the query as shown below:<\/p>\n\n\n\n A couple of tips when working with Athena:<\/p>\n\n\n\n Quicksight can read Athena\u2019s query and present charts and diagrams from them. It\u2019s very straightforward: just go to the Quicksight service\u2019s page and follow one of the many tutorials<\/a> about it, keeping in mind a few important things:<\/p>\n\n\n\n If you don\u2019t want or can\u2019t use Quicksight, you can always call Athena\u2019s API directly and build your own dashboard from data.<\/p>\n\n\n\n The machine learning algorithm we will explore in this article is called Random Cut Forest. The algorithm takes a bunch of random data points (Random), cuts them to the same number of points<\/strong>, and creates trees (Cut). Finally, it checks all the trees together (Forest) to verify if a particular data point is to be considered an anomaly.<\/p>\n\n\n\n Generally speaking, a tree is an ordered way of storing numerical data, and to create it, we randomly subdivide the data points until it is possible to isolate the point we\u2019re testing to determine whether it\u2019s an anomaly. A new level of the tree is created whenever we subdivide the points.<\/p>\n\n\n\n Sagemaker offers a built-in implementation of Random Cut forest which accepts data points in CSV format. We can retrieve them easily with:<\/p>\n\n\n\n Data contains a timestamp<\/strong>, the temperature value<\/strong> in C\u00b0, and aroom_id<\/strong>, which identifies a particular room where the sensor was installed. We have already used our Step Function to divide data coming from different rooms so we can directly pass the CSV to the Estimator.<\/p>\n\n\n\n Sample data extract We referred to this article<\/a> to verify how data must be passed to the Estimator. According to the official documentation, we need to pass 3 main hyperparameters:<\/p>\n\n\n\n The Estimator is defined in this way:<\/p>\n\n\n\n Some considerations to take into account are that we generate the execution_role<\/strong> and thesagemaker_session<\/strong> using the built-in methods. For our training, we use an ml.m4xlarge instanc<\/strong>e, while for inference we used an ml.c5.xlarge<\/strong> as suggested by the docs. Don\u2019t\u2019 waste credits on GPU instances as the RCF algorithm doesn\u2019t take GPU into account.<\/p>\n\n\n\n For deploying we can use the standard approach:<\/p>\n\n\n\n And that\u2019s it! We have reached the end of this workflow. Let\u2019s see some references and sum up all we have seen until now.<\/p>\n\n\n\n In this article, we have seen many services from AWS perfectly suited for data analytics when dealing with near real-time scenarios. We have discussed about AWS Step function and how it can be used to orchestrate nonlinear workflows, giving developers the ability to have multiple choices in manipulating and extracting data for different kinds of analysis.<\/p>\n\n\n\n AWS Glue proved to be flexible enough to take care of data sources residing in different places: EC2 instances, S3, and in different accounts. It was a perfect choice also due to the simplicity of setting up Spark jobs. We have seen in particular how to connect to a data source using a JDBC connection.<\/p>\n\n\n\n Athena demonstrated to be the perfect tool to extract ETL results for Business Intelligence fruition, and Quicksight the obvious choice to show the results, as it\u2019s natively compatible with Athena queries.<\/p>\n\n\n\n As in many other scenarios we have faced, Kinesis Data Firehose was also used to transfer near real-time data to S3 from a non-AWS source.<\/p>\n\n\n\n We have also seen how Amazon S3 is always a must-have when dealing with big data workflows, machine learning problems, and data lake creation. Its durability standards, as well as its being compatible with any other AWS service, makes it the perfect choice both for long term storage and in-between steps buffer.<\/p>\n\n\n\n To conclude we gave some hints on how to manipulate data in SageMaker to carry out inference for anomaly detection.<\/p>\n\n\n\n This concludes our journey for today, as always feel free to comment and reach us to discuss any question, doubt, or idea that comes to your mind. We\u2019ll be glad to respond as soon as possible!<\/p>\n\n\n\n![\"infrastructure](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2021\/02\/image8-1-1024x658.png\")
Kinesis Data Firehose <\/h2>\n\n\n\n
![\"Select](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2021\/02\/image4-1-1024x392.png\")
const deviceModule = require('aws-iot-device-sdk').device;\nconst cmdLineProcess = require('aws-iot-device-sdk\/examples\/lib\/cmdline');\n\u2026\ndevice.publish('topic', JSON.stringify(payload));\n<\/pre>\n\n\n\n{\t\n \u201ctimestamp\u201d: \u201cYYYY-MM-DD HH:MM:SS\u201d,\n \u201croom_id\u201d: \u201cXXXX\u201d,\n \u201ctemperature\u201d: 99\n}\n<\/pre>\n\n\n\n![\"Delivery](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2021\/02\/image7-1-1024x823.png\")
Glue crawlers and connectors<\/h2>\n\n\n\n
![\"JDBC](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2021\/02\/image2-1-956x1024.png\")
\n
![\"Target](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2021\/02\/image1-1-1024x600.png\")
import sys\nfrom awsglue.transforms import *\nfrom awsglue.utils import getResolvedOptions\nfrom pyspark.context import SparkContext\nfrom awsglue.context import GlueContext\nfrom awsglue.job import Job\nfrom awsglue.dynamicframe import DynamicFrame\n<\/pre>\n\n\n\n
args = getResolvedOptions(sys.argv, ['JOB_NAME'])\nsc = SparkContext()\nglueContext = GlueContext(sc)\nspark = glueContext.spark_session\njob = Job(glueContext)\njob.init(args['JOB_NAME'], args)\n<\/pre>\n\n\n\n
rooms_temperatures_df = glueContext.create_dynamic_frame.from_catalog(database = \"raw_temperatures\", table_name = \"temperatures\", transformation_ctx = \"temperature_transforms\").toDF()\nrooms_temperatures_df.createOrReplaceTempView(\"TEMPERATURES\")\n<\/pre>\n\n\n\n
result = glueContext.sql(\"
dynamicFrameResult = DynamicFrame.fromDF(result, glueContext, \"Result\")\napplymapping = ApplyMapping.apply(frame = dynamicFrameResult, mappings = [(\"temp\", \"bigint\", \"temp\",\"bigint\"), (\"room_id\", \"string\", \"room_id\",\"string\"), (\"timestamp\", \"string\", \"timestamp\",\"string\")])\n<\/pre>\n\n\n\n
to_be_written = glueContext.write_dynamic_frame.from_options(frame = applymapping, connection_type = \"s3\", connection_options = {\"path\": \"s3:\/\/Step Function<\/h2>\n\n\n\n
![\"Our](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2021\/02\/image3-1.png\")
\"Type\": \"Parallel\",\n \"Branches\": [\n {\n \"StartAt\": \"Import Raw from EXTERNAL_DB\",\n \"States\": {\n \"Import Raw from EXTERNAL_DB\": {\n \"Type\": \"Task\",\n \"Resource\": \"arn:aws:states:::glue:startJobRun.sync\",\n<\/pre>\n\n\n\n\"ResultPath\": \"$.ParallelExecutionOutput\",\n\"Next\": \"Start LAKE_DATA Crawler\"\n<\/pre>\n\n\n\n
\"Parameters\": {\n \"JobName\": \"EXTERNAL_DB_IMPORT_TO_RAW\",\n \"Arguments\": {\n \"--enable-glue-datacatalog\": \"true\",\n<\/pre>\n\n\n\n\"Catch\": [\n {\n \"ErrorEquals\": [\n \"States.TaskFailed\"\n ],\n \"Next\": \"Data Pipeline Failure\"\n }\n],\n<\/pre>\n\n\n\n\"Wait for LAKE_DATA Crawler\": {\n \"Type\": \"Wait\",\n \"Seconds\": 5,\n \"Next\": \"Check LAKE_DATA Crawler\"\n},\n<\/pre>\n\n\n\nAthena and Quicksight<\/h2>\n\n\n\n
\n
![\"Athena](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2021\/02\/image5-1-1024x312.png\")
\n
\n
Sagemaker: Random Cut Forest anomaly detection<\/h2>\n\n\n\n
data_location = f\u201ds3:\/\/{bucket}\/{key}\u201d\ndf=pd.read_csv(data_location,delimiter=\u2019,\u2019)\n<\/pre>\n\n\n\n![\"Sample](\"https:\/\/blog.besharp.it\/wp-content\/uploads\/2021\/02\/image6-1.png\")
\n
import sagemaker\nfrom sagemaker import RandomCutForest\n \nexecution_role = sagemaker.get_execution_role()\nsagemaker_session = sagemaker.Session()\nbucket = \u201c
rcf.deploy(initial_instance_count=1, instance_type=\"ml.m4.xlarge\")\n<\/pre>\n\n\n\n
References<\/h2>\n\n\n\n
\n
Takeaways<\/h2>\n\n\n\n