Stream IoT sensor data from Azure IoT Hub into Databricks Delta Lake
In Data we trust
IoT devices produce a lot of data very fast. Capturing data from all those devices, which could be at millions, and managing them is the very first step in building a successful and effective IoT platform.
Like any other data solution, an IoT data platform could be built on-premise or on cloud. I’m a huge fan of cloud based solutions specially PaaS offerings. After doing a little bit of research I decided to go with Azure since it has the most comprehensive and easy to use set of service offerings when it comes to IoT and they are reasonably priced. In this post, I am going to show how to build the architecture displayed in the diagram below: connect your devices to Azure IoT Hub and then ingest records into Databricks Delta Lake as they stream in using Spark Streaming.
The very first step is to set up Azure IoT Hub, register a device with it and test it by sending data across. This is very well explained by Microsoft here . Make sure you follow all the steps and you’re able to read the messages sent to IoT Hub at the end.
The only extra step we need to take is to add a new consumer group to the IoT Hub. Doing this means our Spark Streaming application will have its own offset, tracking where in the queue it has last read the records coming from devices. By assigning unique consumer groups to each application that subscribes to IoT Hub, we can send the record coming from IoT devices to multiple destinations, for example to store them in Blob storage, send them to Azure Stream Analytics and do real-time analytics, as well as a delta table in Databricks Delta Lake.
Navigate to IoT Hub page on the Azure portal and select your hub. Click on Built-in endpoints and add a new Consumer Group:
Moving on to the next layer in our architecture, we’re going to set up Databricks. Databricks offers a platform that unifies data engineering, data science and business logic. It is basically PaaS offering for Spark on cloud, which speeds up data exploration and preparation.
Delta Lake is a storage layer invented by Databricks to bring ACID transactions to big data workloads. This is a response to limitation within an existing big data storage mechanisms like Parquet: They are immutable. To update a record within a Parquet file, you need to re-write the whole file. With Delta, you can easily write update statements at records level. This is all we need to know about Delta file format for the purpose of what we want to build here, more about is here .
A very important result of this feature for IoT and streaming use cases is that we will be able to query the data as they arrive, instead of having to wait for a partition to be updated (re-written)
In this solution we will see how to set up Databricks, use Spark Streaming to subscribe to records coming in to Azure IoT Hub, and write them to a Delta table.
Navigate to Azure Portal and click on Create a Resource -> Analytics -> Azure Databricks. This is where you create a workspace, which is where you can access all your databricks assets. Fill up the new form that opens up and make sure you select Standard for pricing tier. Then hit Create:
When the workspace is created, go to Azure Databricks Workspace resource page and click on Lunch Workspace. You will be navigated to your workspace. Create a new cluster with the same properties you see in the picture below. You can ask for bigger nodes or enable autoscaling, but it’s not needed for this tutorial:
The next step is to create a notebook. Click on Home -> <Your Email Address> -> Create -> Notebook. Give it a name, select Scala as the default language of the notebook (you can change it later using %), and select the cluster where this notebook’s commands will run on.
Now that we have setup our notebook, we can start writing code.
The code snippet above first creates a connection string pointing to the IoT Hub we created before. The only extra steps you need to take is to get the connection string from Azure portal and replace it in ConnectionStringBuilder and change the name in .setEventHubName to “<Event Hub-compatible name>” accordingly. Open Azure portal and go to your IoT Hub’s page. Click on Built-in endpoints and copy what you see below and paste in the code snippet in the notebook:
What we get after those commands are completed successfully is a DataFrame that has the following fields in it. The messages coming from our IoT device are in the “body” field:
The next step would be to extract the device data coming in the body field of the DataFrame we built in previous step and build the DataFrame comprising of the fields we want to store in our Delta Lake to do analytics later on:
The resulting DataFrame looks like:
Now we can create a table from our DataFrame and start writing SQL commands on it:
We’re almost there. We have the data we receive from our IoT device in a Spark SQL table, which enables us to transform it easily with SQL commands.
Tables in a Big Data ecosystem are supposed to be partitioned. I mean they better be, otherwise they’ll cause all sorts of problems. The reason I extracted Enqueued_Time from JSON was to be able to partition my table by date/hour. IoT devices produce a lot of data and partitioning them by hour not only makes each partition reasonably sized, but also enable a certain type of analytics to be performed on the data when companies need to predict the performance of their devices at different times of the day or night, for example.
The resulting DataFrame has the following schema:
The final step is to write the stream to a Delta table:
Let’s check the options passed to writeStream:
outputMode: Specifies how the records of a streaming DataFrame are written to the streaming sink. There are 2 modes: Append: Only the new records will be written to the sink Complete: All records will be written to the sink every time there is an update Update: Only the updated records will be outputed to sink
Append: Only the new records will be written to the sink
Complete: All records will be written to the sink every time there is an update
Update: Only the updated records will be outputed to sink
option: checkpointLocation This is needed to ensure fault-tolerance. Basically we specify a location to save all application progress information. This is specially important in case of a Driver failure, read more here .
This is needed to ensure fault-tolerance. Basically we specify a location to save all application progress information. This is specially important in case of a Driver failure, read more here .
format: The output sink where the result will be written, obviously “delta”.
partitionBy: The column(s) by which we want our table to be partitioned by. We decided to partition our table hourly as explained above, so we pass in date and hour.
table: The name of the table.
If all the steps above have worked, you should be able to query your table and see the records inserted into the Delta table by running the following command:
And we’re done! Now we have a table in our Delta Lake that holds our IoT devices data. You can take it from here and do ML on the data collected or mix it with other tables you might have in your Delta Lake. And definitely feel free to ask your questions below in comments section.
Tagged
Azure
Databricks
Delta Lake
Internet of Things (IoT)
IOT
IoT Hub
Spark Streaming
Big and streaming data enthusiast based in Melbourne, Australia and travelling throughout Asia View all posts by Saeed Barghi
Fill in your details below or click an icon to log in:
You are commenting using your WordPress.com account. ( Log Out / Change )
You are commenting using your Google account. ( Log Out / Change )
You are commenting using your Twitter account. ( Log Out / Change )
You are commenting using your Facebook account. ( Log Out / Change )
Connecting to %s
Notify me of new comments via email.
Notify me of new posts via email.
June 2019 (1)
April 2019 (1)
February 2019 (1)
January 2019 (1)
May 2018 (1)
April 2018 (2)
January 2018 (1)
December 2017 (1)
October 2017 (1)
September 2017 (1)
August 2017 (1)
June 2016 (2)
May 2016 (1)
July 2015 (1)
February 2015 (2)
May 2014 (1)
April 2014 (1)
June 2013 (1)
March 2012 (1)
September 2011 (1)
August 2011 (3)
June 2011 (1)
March 2011 (1)
February 2011 (2)
January 2011 (2)
Enter your email address to subscribe to this blog and receive notifications of new posts by email.
Join 29 other followers
Sign me up!