Scale By The Bay 2019: Sharanya Santhanam & Shiv Toolsidass

Scale By The Bay 2019: Sharanya Santhanam & Shiv Toolsidass

Recording: Scale By The Bay 2019: Sharanya Santhanam & Shiv Toolsidass

[Music] hi good morning everyone I'm sure Anya I woke in the data infrastructure team at lift so the agenda for today is we're going to talk about enabling query real time data with druid link and Kafka so the agenda for today the motivation why do we really need this what our architecture was before a quick peek into what druid is and why we decided to go with druid a deep dive into our new real-time data pipeline a cluster set up overview and how druid and all of these new pieces fit into our platform and we talked about a few of the use cases we have so talking about motivation so lift is a very data heavy company and real time data is really crucial for our business today for example we want to track what is market help at as a for airport and we typically have general managers in charge of certain regions and all those real-time data really helps them because they have access to a lot of levers that they can pull to shape supply and demand and secondly geospatial data is core to our business we need to track our riders and drivers in our various regions so the need for to be able to query in real time is super important for us I have a slide on the architecture but I can just talk about it so we have right on the left on the top we have our mobile apps and micro services which basically generate our event data in a JSON format and that lands on to a Kinesis stream and there is an application that leads from the skinny's stream and dumps it into s3 in parky format we have we use hive as our ETL engine and a lot of our hive ETLs are orchestrated from airflow tags so every night we basically read these s3 files in RAW format generate these park' files and a lot of the downstream ETL is kickin to create fact and dimension tables once all of these tables are generated the data is available to query in presto and visualizations through superset so with the current architecture what was really missing is two pieces one is there was a 24-hour lag so anybody querying by a presto couldn't see what really happened in the last 24 hours and secondly presto as a query engine it's very powerful but p75 is typically in like 30 seconds that works really great for engineers but not for someone and maybe an Operations team who wants to make really quick decisions based on real time data they need a really snappy query engine so that was missing so what we were really looking for is data freshness under a minute a p95 less than five seconds and a good inbuilt geospatial support so here comes druid there's a lot of texture but I just copy this from the open source web site but TLDR version druid is an in-memory column-oriented data store it's a query engine and a database designed for time series data and has a lot of features for Search Indexing so what that really enables is very fast data exploration with sub-second query latencies and druid is known to scale at a lot of other companies like eBay Airbnb etc so why did we go and Drew it this actually captures a lot of the features that Drew it provides so druid supports native time indexing so if you basically think of it like every row for the sake of simplicity would have a time stamp on it and a user can the producer of this data can then dictate at what granularity they want to store the data the time stamp like at the second level minute level whatnot and Druid categorizes its columns as either dimensions or metrics so dimensions is basically columns on which you could potentially do a filter or a group buy on so if you look at lift data like this is like an example of our rides data set so the dimensions would be what kind of a ride is this it is the line ride is it a classic ride what is the status of the ride and metrics are columns which are basically aggregates and these are computed as the data is getting ingested into Druid and you can also have aggregation roll-up that can take place at ingestion time but that is totally optional so your metrics would be something like what is the average write duration for example in addition to the schema druid has its own internal data format that's known as segments and it's highly optimized it creates indexes with using a lot of bitmaps and reverse indexes on all of the dimension columns which really helps speed up lookup queries scan queries as well as time series queries and this data is uses lc4 compression and dictionary encoding for a lot of the data structure so it's very compact and cost-effective second thing is Druid stores all of its data in memory so when you actually submit a query the aggregate push down is actually pushed on to each of the serving nodes there's no shuffling of data between the serving notes and the result set is just returned back to a broken old which does the final merge and sends the result set back to the user and the brokers also make use of direct memory Maps for a lot of the merging so that really helps speed up queries especially boosts group by queries by a great extent a lot of most of all of the components actually of the druid architecture are highly available and horizontally scalable so it's very easy to quickly auto scale and tune up as you need grows and another thing that's really nice about druid is it out-of-the-box supports both batch and real-time ingestion and we make use of this at lift so for batch we join a lot of our hive tables from our data warehouse and submit a batch job which internally kicks off a MapReduce job and in the reduce phase it produces these druid segments the real-time pipeline basically works out as do it being a Kafka consumer we have a fling application that processes our events data and dumps it on to a Kafka topic and Drouet starts ingesting this in real time and the data is available to query within seconds another really nice thing about druid which helped us gravitate towards it was it has very clean API is easy to submit queries through both JSON as well as it also a sequel support so easy to integrate with our existing infrastructure and we use super set quite heavily here at lift what's nice is you don't have to write any sequel with super set an Operations person for example can just quickly start visualizing creating charts and dashboards off of do it without writing a single piece of secret gold and druid is if you look at the page the community page they have a lot of community extensions and plugins contributed so that really helps you create a flavor of druid custom to your needs we make a use of make use of a bunch of these custom extensions so this is our real time pipeline we still have the services on the top left I mean in the left which is the mobile app and the micro server is generating these events which make it into Kinesis we've introduced a new blink application flink is a streaming application and what's nice about slink is it has very good support for functional api's as well as sequel so someone could just write a sequel query to transform the event data and the flink app would then apply those transformations onto the data you can also configure singles and sources and plink so with the help of that we use sequel as well as some custom UDF's for geo hash transformations etc and we dumped this transform data onto a Kafka cluster alter topic in proto format so the reason we use proto is druid ingestion it's supposed to be a lot more performant with either Avro or proto versus JSON because of all the the schema that's built in to the data and the type safety that comes with it at lift we also make use of an ideal repo which is basically proto schema definition for all of our events and we cannot be back on this at the various levels at the various levels of the data pipeline to do schema validation checks we've also customized druid to have a custom row parser where the D serialization can happen just using dynamic messages dynamic proto messages and since it's proto and druid also has very nice graceful handoff which enables us to do graceful scheme lucien cool that's it I'm gonna handle the shoe hey everyone so I'm gonna chat a little bit about our custom setup for joy deadlift so we use a fork of the open-source project and every time we make a change in deploy code this generates a debian package that's shipped out to every rural instance that lifts something we found to be really good while you're setting up jus it is that the open source documentation provides really good pointers on how to set up your cluster what sort of instances you might want to choose based on which clouds or cloud provider you're using so I'd lift we have a jewel cluster that stores about tens of terabytes of data we process hundreds of thousands of records a second and you can see the breakdown of the AWS instances would that be used for each of our druid processes so a little trick that we found to be useful is that druid has two kinds of processes called historical and middle managers we found that when you call ok the same processes on the same nodes you end up you know having to have less instances overall which leads to a lot of cost savings for us we also have a fling cluster a Kafka cluster and to store metadata druid uses zookeeper and my sequel RDS and the final piece is s3 which we use for deep storage so like you know you can clearly see that the setup is pretty complex and it has a lot of moving parts which operationally if you go on call there's a lot that you need to know about so I think this is one of the few cons that we ran into a setting up drew a deadlift so Sharon you talked about this before but I'm going to go into more detail we run a nightly batch pipeline every night this batch pipeline works using Apache airflow and Hadoop and the reason that this ends up being really useful for us is that it enables three things one is compaction of segments so the real-time pipeline during ingestion and build segments that are not really optimized for storage but the the batch pipeline you get optimal compression for your segments so we basically run this batch battle on the same deer that's ingested in the real-time pipeline so that it compacts the segments and that allows us really good cost savings because Druid stores its data in memory and memory is super expensive so another advantage of this process is that when you ingest data at such a high throughput real-time it ends up being a little bit lossy so this batch pipeline ends up filling all those gaps in for us and the last reason it's super easy was if there's any downtime in the real-time pipeline this serves as a fallback for us so that consumers don't lose data so here's how a product team or any customer of Druid can onboard we ask them to define flank sequel which is the transformation they want performed on their event stream so you can see that in step one a user then defines a protobuf definition which is exactly how they want their table in in Druid the third step is some custom tooling we built at lift which aims to make it as simple as possible for any user who's unfamiliar with Truett to onboard so what they do is they we ask them to define granularity of their segments which is true it's unit of storage we ask them you know what kind of granularity they anticipate having when they query their data and we also ask them to define metrics and dimensions and which columns correspond to their timestamps in the fourth step you can see that this is actually a manual command of that we have in the in our terminal but this is triggered every time you deploy code to do it so what this does and under the hood is that it will go and perform checks against our schema registry and send this configuration file in the form of druids configuration file to Druitt so it basically parses this configuration file and creates a custom config for druid so we found three distinct use cases to be really useful at lift one is web applications the second is the other exploration and analysis and debugging and the third is using druid as a time series the database so let's have a look at case one interactive web applications so I've left we have a lot of geospatial data because we're a rideshare company and our teams need a way to query this geospatial data in real time so a user defines a time window they define their little box which is there which corresponds to a geo hash and they want to do certain lookups on their data our teams are left explored using transactional data stores for this use case but what they found is that query latency is for time series data ends up being really high and also transactional data stores are pretty expensive so the reasons that druid you know really stood out for this use case is that one left is a Python shop so a lot of our back-end services use Python and pi drew it's an open source library that basically allows us to query druid data from any lip-service druid can as we mentioned earlier in just hundreds of thousands of records a second and it offers query latencies of around 100 milliseconds and with our real-time pipeline set up our data freshness is around less than 1 minute so and the last bit is that you know using this dashboard and users don't have to worry about writing sequel they can just you know create a little bounding box and figure out you know whatever data corresponds to that those coordinates so we have a ton of users that like have interactive web applications built on top of druid just like this for use cases that are not geospatial data it the second case that we found to be really useful is debugging network traffic Oh sir sorry it's basically data exploration and an analysis like shurni i mentioned earlier we have apache superset which we use as our visualization tool at lift and a specific example of this is you know teams using these dashboards to debug network traffic so the fuses on where's its networking mesh and each request - on what generates a corresponding log and we ingest a subset of these logs into Drouet so that users can you know figure out at a very high level what kind of traffic their services are getting so with druids data freshness under one minute a query latency of around one second you teams can debug network traffic in real time and this ends up being super useful especially during something like a DDoS attack when you need to figure out you know what your network headers your source of your traffic is so like this we have a lot of other dashboards built out that where you drew in and the third use keys that we found to be really useful for druid is generally as the time to use database so a really cool use case for for this was our experimentation team needed a way to tell users who configure experiments how their experiments actually you know perform and how they affect our key business metrics like rights taken rights dropped off so the experimentation team also wanted to send an alert to a user as soon as you know we figure out that an experiment is not going the way it's planned right because if you have a bad experiment that can potentially lead to an outage so once again with our data freshness being under a minute and with p99 correlated in seas around a second we ended up being able to build an alerting system on top of druid so another thing that we we want to point out is that druid can't perform joins against against tables so and the reason it does this it can do this is because it offers certain query latency guarantees and when you perform joins it can't really stick to those guarantees and also yeah so to basically get around the fact that druid can perform joins we ended up performing joins at the streaming layer and as journey I mentioned earlier we use Apache Fling which is a stream processing engine and so Apache flink is where we actually perform a joint across multiple streams and doing this actually allowed us to you know gain insights into how one stream of events affects the other one and this this tool is currently being used in production today and it's very useful to debug how your experiment affects key business metrics so yeah that's all yeah thank you so much for coming out for a talk [Applause] [Music]