data.bythebay.io: Dinesh Narayanan, Building realtime efficient queries in an IOT platform
good afternoon everyone thanks for coming um in this talk I'm going to briefly describe some of the patterns that you may want to consider when building your realtime or batch processing pipelines uh quick one minute introduction about who we are uh we are an iot based platform uh what this allows you to do is that your devices can talk to other devices in a secure way and this is being done in a manner such that it's independent of the underlying Wireless or any of the transmission protocol um we as far as the injection goes we support various kinds of protocols being uh HTTP web saret uh mqt or Co let's look at a couple of examples uh to see which protocol might make sense say I have a smartwatch and uh and say it has a bunch of sensors embedded in it uh such a device uh may be collecting data points related to say heart rate ECG or PPG or maybe your skin temperature it could be any of your body metrics uh such a device could be generating data points that is of the order of 100,000 Plus data points in a very short period of time uh you would want to use web socket in that case uh if I have a device that has low processing uh Power uh you would want to use coab you know maybe it's a microcontroller um and as far as the injection goes um we collect uh various kinds of data be uh you know be uh historical or you know continuous or discret we have a powerful rooll engine to aument realtime alerts and actions um I will go the text text tack in my next slide uh but it enables you uh to send Real Time alerts the pain point right now in the iot space is that uh a device cannot talk to the other device in a secure way uh we are trying to solve that problem uh and hopefully everything goes fine um this is a quick slide and if you want to learn more about rtic Cloud uh you know you can go to Artic darkcloud uh this is about me um a quick slide about our uh backend stack we are a jvm shop um most of our code base is in Scala um we use Apachi misos uh for deploying most of our stateless Services uh through Marathon um our micro services are based on play framework uh we use MySQL for managing our entities um we use Apachi Cassandra to manage all the time series data and we argument that with elastic search to answer some of the dynamic queries the challenge here is uh how much you duplicate in Cassandra versus you know you go in elastic search so you have to make some choices there in terms of performance uh something to consider U messages within our pipeline um are mostly encoded in AO or message AG pack format so that you know uh you spend less time on serialization um we use Apachi par for storing structured data in hdfs uh our bad jobs are mostly written in Apachi spark uh as far as the stream processing Frameworks we use uh samza and Spark streaming um we are in the process of evaluating Apache Flink uh we have some computations and Transformations that's based on event time and U we want to use Apache Flink for that use case uh our rule engine is based on samza AA cluster and AA streams uh for data sets that fit in memory we use redif um I'm going to talk about uh since this is a 20-minute talk uh I will briefly touch up on a couple of Frameworks that uh we think we like it um the first one that I want to talk about is Apachi par Apachi par is a columnar storage format that enables greater efficiency in terms of file size and query performance uh it's originally based on Google's rumel paper um the advantage you get is that that your file size uh is uh much smaller than what the row oriented equivalence would be uh if you are dealing with hundreds of terabytes of data sets uh this adds a huge value here it could be you know even 30 to 40% of savings uh when you're running your software on AWS Cloud this does matter um it does uh now let's discuss how it's able to achieve the uh storage savings um it does this by storing The Columns next to each other which allows very efficient encoding techniques the first encoding that it applies is the dictionary encoding dictionary encoding what it does is it takes your arbitrary Long Key text and encodes them into a small value small in uh thereby saving storage uh Delta encoding can be used for temporal data wherein it is sufficient to just store the first entry and the values from 2 to n uh it's sufficient to store the differences uh you can use r when you have continuous duplicate data wherein it is sufficient to store the first value and uni the first unique value and followed by the number of duplicates um by applying all these encoding techniques um you realize Savings in storage uh to convey the point um say you have your uh actual records I guess I need to be closer um um say this this uh the first table list the actual records and your row representation uh it would uh each record would be placed one after the other uh in a columnar representation uh The Columns are placed next to each other now the data is homogeneous and you can apply these efficient encoding techniques it also makes your data more cash friendly there is less misprediction in terms of branches so your jobs will run much faster than the traditional sequential uh sequence file counter paaths um the cool thing about Apache par is that um it allows you to store deeply nested structures in a columnar fashion what this means is that your existing applications can continue to use AO Thrift or maybe raw Json uh to send uh to send the data inbound I highly recommend you use a serialization format such as a or Thrift uh so that you know you save Less on serialization cost um if you use Apache AO you have the advantage that your data on the schema can evolve in a Backward Compatible way what this means is that uh hey I made a schema change and now I broke my bad jobs you wouldn't want that you would like to know the failure right before say you youqing the event um so so far we talked about the uh storage savings uh the next advantage that you get is your query performance with Apachi par the query engine can and skip over columns that are not needed to answer a query uh it does this by uh pushing your predicate and projection push terms all the way down to the storage layer and thereby saving on iio cost and um here is an example um say I have I'm collecting a body metric and this is an AOS schema um say I have device ID and device type which I have marked it as mandatory uh ECG and uh p pg2 are optional uh the device might or might not emit them and channel is mandatory here uh so this is the road representation and now the columnar equivalent is that is as follows the syntax is very similar to uh Proto buff uh things that are been marked mandatory here say device ID or device type uh there is this prefix called required and for optional ones like ECG you know there is this uh you need to put in that optional keyword uh string is as you have seen here string is a logical type in Apachi par the internal representation is that it's stored as a binary with utf8 encoding um so this example uh shows us a schema without any nesting the next example um I hope you can see my screen um I can see here um this is a this is a bit more involved U wherein we are uh showing examples of how to use the how to use collection or container data types like list or map um again device ID device type uh those are mandatory Fields um the TXS U you know uh repeated means there can be zero or one occurrence which means it's optional and to encode a list you put it within a group what that means is that uh now my measurement can have an array of values say for channel ECG I could I could pack in a bunch of floats um now introducing Apachi Park means that you have an extra face in your pipeline uh whose primary requirement is to uh convert the row representation into your columnar representation uh here is a sample workflow uh for how you would build such a pipeline uh you can uh the first step is the even still uh sending data in the original format which is a row oriented style you know could be AO or plain J song uh and say you have a micros service you know that's receiving those events which does the validation normalization or you know or any other transformation and once the transformation is complete you can queue it to a CFA topic uh with there is a new library from Kafka guys which allows called schema registry um which allows you to validate any message that sent in AO format so uh before queuing it itself you know you can know the errors then and there and once the events are cued now you can have a flume agent uh all it does is it listens to the events real time uh and then it it can write the data onto htfs uh you could uh replace flu magent with Kafka connect here um the advantage with Kafka connect is that your rights the syn side is exactly one semantics uh something you may want to consider and once you have the uh serialized AO data or the raw data onto htfs you can have an MR converted job which can uh which will convert the row representation onto a a columnar representation uh you could alternatively replace this Mr job with a spark job as well it's up to you and once you have the data in columnar representation um you can run your hly or uh daily spark jobs or maybe a different job you know based on whatever threshold it's supposed to run and persist on your sync of choice could be Cassandra or you know writing back to hdfs um as as I said before uh the important factor here is not only just the storage savings but also the query performance um based on our test uh for some of the uh bad jobs we found that the performance in appr by 3 to 4X than running through the uh sequence file based formats um I guess I can skip this slide uh I was trying to convey that uh in a batch model if a job fails you can replay on its entirety so if exactly one semantic is critical and latency is acceptable uh yes a bad job based pipeline would be a good fit um the second section that I want to talk about was a new future that was introduced in the recent versions of Cassandra uh it's the future is called sassy sassy stands for S stable attached secondary index what this allows you to do is that previously you couldn't uh your Cassandra database didn't support didn't have a rich enough support for uh compound queries uh so you had to rely on elastic search in the past now Cassandra can do that job for you uh let's look at some of the examples uh that was previously possible in the earlier versions of Cassandra um the first uh feature was that you can do prefix matches say I can get all messages or events uh for the PPG Channel um you can do contents class which allows you to do infix based searches um you can plug in different types of analyzers out of the box Cassandra has support for non- tokenizing analyzer if you want to do a term based search and then you can also plug in a standard analyzer if you use a standard analyzer you can uh put in things like uh hey what loc you're targeting whether you want to enable stemming whether the search should be case sensitive or not and if you want to discard stop wordss all of this functionality can now be done in Cassandra starting version 3.4 U lastly you could do sparse queries sparse queries makes sense when you have unique value say uh on on a tempo data um you can now do range and equality searches as far as the life cycle of The Sassy uh index metadata files it follows very similar to that of kandas ss table what this means is hey my client center right with say consistency level of local one uh the data gets uh uh the replica which or the primary which is supposed to receive the data it will write it onto the commit log then it'll write it onto the M table and then send an acknowledgement back to the uh uh back to the client or it may sent to the coordinator if you use a toon a policy uh the client gets the confirmation so there is a you can there's a one less Network hop that you can avoid by using token policy um now if there's a sassy index associated with the Cassandra column family it does an extra step of maintaining your inmemory index structure and when the M table gets full what it does is the data that's in M table gets flushed onto SS table and similarly the data that's in the inmemory index structure gets flushed onto a SS sassy index file so you have a one to1 correspondence between um SS Cassandra SS table and Sassy index file so now so the life cycle is just the same as I said and now during compaction U what happens is that you know when multiple SS table files get merged uh the associated sassy index files also get merged um so it's pretty cool feature uh we haven't we are yet to use this in production but uh we have tested it on a uh on a fairly uh good amount of load and we liked it and most likely we'll be deploying this in BR um there are a few caveats that you need to consider when using sassy and this applies in General to um secondary indexes as well um when you use secondary index in Cassandra it is very important you provide um partition key if you don't provide a partition key the query will be submitted to every node in your cluster and you can't scale linear anymore uh same caveats apply to U apply to sassy index as well um there is no support for sorting um if you are working on a Time series data and uh and you need sorting because the users would like to paginate real time through some kind of an API you have to move the Sorting onto the uh as a clustering key uh there is no support for scoring um the third one which I think I found uh interesting is that uh it requires two passes to fetch your data uh first pass to read the index file and second pass to uh read the actual data so search engines have a bit of an advantage here and if you have very wide partitions then sassy only maintains the partition offset from there on it has to do a linear search uh some C that's worth considering um okay I guess uh more of time just want to some question okay uh I was I was just going to conclude that uh we have a couple of examples of how you would use Sassy index uh the first one demonstrates how you would create a prefix based search using a non- tokenizing analyzer so you can can do uh searches like this guy here the prefix match and the second example allows you to do infix based search uh and you have the option to specify your analyzer and if you want to enable stemming or you know if you want to enable you know your local or skip stop words and how you want to normalize uh the data that you're indexing uh I guess I'm not going to have time to discuss uh streaming Primitives uh but but uh the takeaway is that you can now uh react in a very low latency way okay I guess I will stop here and take time for questions um I had a question about the streaming so if I'm an application on the cloudform I then what are the ways in um is a um can you repeat the question so if if I'm building an application that requires streaming events from a device say other devices what are the options for consuming that as application so um the question is what options you can consider when building your stream processing uh from an iot perspective um so today we have uh Apachi spark streaming or you can do use Apachi sza or you can use uh Apache Flink um spark streaming as a version 1.6 does not have support for event time processing uh they are going to add that future in 2.0 uh but then uh when you are window based on uh event time you need the notion of completeness uh and they haven't yet documented how they are going to handle messages which arrive uh late in the pipeline um so that you know you still uh emit results in a speculative manner Flink has out of the box step support right now uh I would say Flink has a bit of a momentum right now because of the rich Primitives which it built based on Google's data [Music] flow yeah I was originally a Java developer um scalar does take some time to learn U but uh we found that uh there is definitely some Savings in terms of code size uh the test libraries like scalar test or awesome and uh you know and even as far as the Frameworks play is pretty decent uh finagle is even better so uh so the as far as the libraries are concerned um it's fairly rich and you have some very good options that you want to cons thanks okay [Applause] cool