How Razorpay refreshes its data warehouse 10x faster with graphs and indexes

Uh this is Amit Prabhu. Uh I work as a staff engineer in the core data platform team at Razer Pay. Sure. Slightly a mic problem. One second. Sure. So my team at uh Razer Pay uh we work on building a central data layer uh which ingests data from multiple sources and then uh our job is to make sure that the data is easily accessible by both internal and the external users. So today I'm here to talk about one such interesting problem. So this is about serving the transactional reports to our end users. And if you look at the problem, it sounds simple that you are just doing a report generation. But when you add the constraints of uh scalability uh you have data freshness, cost and other things. This becomes more and more challenging. So our we tried multiple solutions and we arrived at a final solution which was 10x faster, 10x cheaper and at the same time we were also able to give better experience to our users. So this is what I want to cover in this talk today and let's get started. So before I uh jump onto the problem statement, I want to give a glimpse about what Razer Pay does and what are what is the uh numbers that we operate in. So Razer Pay is a fullstack uh financial service company. So we uh primarily provide a payment gateway service to our businesses uh which merchants can use to collect money from the customers and they can also do disbburus. So we have a option of payouts. So where they can actually send out cash backs, refunds and other things. So now we have around uh 10 million plus merchants registered with Razer Way across the world and they make around 20 million plus transactions on a single day and they generate around 1 million plus reports uh in a given month and there are about 100 plus microservices. So we have one micros service for each entity that comes in your payment life cycle and then uh they make about 6 billion plus events. So these events include your application events and then uh change data capture events those which come from your database to uh Kafka and together they make around 50 plus GB of data that is ingested every day every single day uh on our data lakehouse. So before uh so now I want to go into the problem of report generation and in order to understand that journey we have to go back to the initial years of Razer Pay. So this is uh uh a couple of years back when Razer Pay was m making around 1 to three lakh transactions per day. So uh we had one big monolithic service that had all the modules built in. It was doing the complete payment processing and it was writing all the data to a single MySQL cluster and reporting was plain simple back then. So we just had to join couple of tables and then create a CSV or a Excel report. So if you look at the slide, so there is this sample settlement report that we generate. It has all the transactions that were done for a merchant and it includes the details about what was the payment method uh how much tax did we charge and uh whether that payment was settled or not. So these kind of details uh that we capture in a transactional report and u merchant use this reports for doing the reconciliation they also run some uh critical workflow on their side. So few example I can give you is we have AMC customers who use reports for stock allocation because they want a confirmation that the money is actually settled to the account and also we have uh quickcommerce customers who use reports for uh dispersing the payouts to their delivery partners. So the reports are actually part of the critical section for some of our merchants and as Razerp started gaining popularity uh Razerp started processing around 1 to 2 million transaction per day. So the scale increased and the obvious choice was to decompose the monolithic service into multiple microservices. So we created one service for each different payment method and different actions that come in the payment life cycle. So from the micros service design every service owns its own database. So it has it has an option that it can choose whether MySQL or Postgress and uh it has the complete autonomy over the database. So earlier if you look at the previous solution we had all the data into a single MySQL cluster. Now it is spread across multiple database clusters and multiple database types as well. So report generation was impossible with multiple clusters. So then we chose a popular architectural pattern that we created data pipelines. So this would look for the change events on the microser tables and then it would replicate the data into a central database. So here we chose tidb as our central database I think which sunny covered in the previous session. uh and we would put all the databases into the central database uh database tables and then reporting service would just do a couple of joins on the TIDB and generate a report. So this was also simple from the reporting point of view and then the scale further increased. Uh now Razer Pay was supporting uh transactions across different countries, different regions and then we started processing around 10 million plus transactions per day. Now the the problem was uh we were keeping all the data on TIDB and this turned out to be a very expensive solution. We were having a self-hosted TIDB and keeping like TBs of data on TIDB was turning out expensive for us. So that's when we wanted an alternative storage which is cheaper to maintain and uh which can support our future scale. So that's when we uh introduced uh data lake on S3 and we chose delta as the table format. Now the data pipelines would replicate the data at two places. One to your TIDB and another to data lake and it would keep the recent data on TIDB and the full data on the data lake. But this still created problems for reporting service because we were not able to generate the complete report from uh data lake primarily for two reasons. So one is the replication latency of 1 to two hours. So we were having a freshness gap of 2 hours on the data lake and the merchants who were generating the report they wanted a real-time data in the report and the next is the query performance. So if you have done any joins on data lake so they are super expensive and you need a very large cluster uh if if your data size is also huge. So one um problematic uh challenge for us us was consider an example you are joining payment and order details for a merchant and this order could have happened two years back and you are making recurring payments on that order. So now you don't know in which date or the month this order falls to. So you cannot apply filters on the order table. So you can just apply filters on the payments table and that means that you are doing a full table scan on the orders. So since the joints were resulting in full table scan on data lake so the queries were going for long hours and they were timing out and that's when we decided that we will serve the recent data which require real-time data from the TIDB and the historical reports would go to the data lake. So this would uh this we did by increasing the SLA for the historical reports but even then we were not able to keep up with the 1 hour SLA for report generation and we wanted a some some other solution and that's when we thought if joins are problematic on data lake why not just premputee the joints. So this is a typical architectural pattern used in industry where you can create a wide table by pre-joining the tables and when you are reading the data from the table you can just do filtering and no need to do any joints and we call this denormalized table as fats and this is how we generate the fats. Uh we have an airflow schedule that runs at off peak covers it triggers a spark batch job on EMR and this job reads all the tables from lake. So we have a configuration file on S3. So this fi configuration file contains which tables to read and which tables to join. So we have one configuration file for each unique report template and uh once it gets all the data from the lake tables uh the bad job would join them and create one large denormalized table and that that's what you see there which which has a large number of columns and it's a very wide table which has all the precomputed joints and we chose delta delta as the table format here. So this worked for some time when the data was less and then we started hitting the challenges. So we were eventually reached to a state where uh we had around 50 plus denormalized facts. uh and this represent different query pattern in the report queries and uh a single fact was joining around 5 to 50 tables and it had hundreds of columns and uh the size of those facts were going above TBS and then the problem that uh generating these facts on every day with such a long execution time was hurting us in terms of cost. So we chose to uh reschedule some of the fats on alternative days and because of that the data freshness increased to 48 hours. So there were three core challenges with the solution. So one is we were doing lot of full table scans to generate the facts and that was resulting in a longunning job which was which were running like for 15 hours and if I do a cumulative count it was taking 100 plus hours per day and uh since these are long longunning jobs so it is clear that you have to deal with spot loss because you won't be running such a long job on a on demand nodes which is very expensive so we went with spot nodes and we had to deal with spot losses so spark would usually recover from spot loss, but sometimes it would crash and it would have to uh restart from the scratch. And since the leg joints were expensive, we had to put every down table into the fact because doing the joins on the runtime for any table was expensive. So we moved everything to the fact and that's why we reached a join of 50 tables that is because to cover all different queries on reports. So we tried multiple short-term approaches and the first long-term attempt we did is creating a streaming solution. So we implemented a spark streaming job. So what it does is it will subscribe to the CDC topics of the tables it wants to join and once it gets all the events it will do a stream join. During the stream join since we have backdated references you might find all you might not find all the references in the streaming window. So that's when we use TIDB as the lookup store. We do the lookups on Tidb to complete the joints and then create one big denormalized uh data frame and then we upsert it on the target table. So this solution also worked for an year and then failed because uh primarily three reasons. One is the mutability of the data. So we had data highly mutable. A payment would go through a a lot many status changes and every uh couple of hour there would be some or the other attribute would change and we had to move that to the uh final fact because reports need accurate data. So we cannot compromise on the accuracy in the reports. So we were dealing with lot of uh events which is 10 million plus events in a 30-inut batch and uh we were doing lot of recomputation again and again and these events were these stream joints were stateful because we were doing lot of lookups and hence uh this created problems for us. The next is we were keeping all the data on TIDV because since we are doing the lookups uh we need references to be present in the TIDv so that we can complete the joints and that was also costing us a lot. And the last problem is the right amplification. So consider an example of a payment link. So using one payment link you can make uh thousands and even lakhs of payments. So now if you use your payment link as a secondary table in your fact and some attribute of that payment link changes now you have to go to uh thousands of payments in that fact and update that attribute. So that is like a right amplification where a one change in payment link is resulting in around lacks lacks of rights on the target team. So these were uh creating problems for us and we eventually discarded the solution. So then we uh went back to the scratchboard and uh brainstormed on how we can incrementally process uh changes on a denormalized table. So one insight was that so one we can create a one-time back fill of the full uh denormalized table and then as and when the change happens we batch the events and then propagate the changes to the target table. But this this sounds simple to understand but how to do that was a problematic. So there there were three questions we need to answer. So one is how do you get the change events uh which you can batch and process. So anyway we were getting the CDC events on Kafka and we wrote a data pipeline that would read this Kafka CDC changes and write to a table on lake partitioned on updated date. So if I have to get the change events on a particular day, I can go to that partition and get all the change events on that day. Now uh change events were sorted. The next part is how do you know which are all the related data? So a change event comes on payment. Now how do you know which all data you need so that you can update the fact. So that's when we created a dependency graph. This is just a inmemory data structure. This does not hold any data. This just tells us that uh in a fact configuration these are the tables I need to join and these are uh related in this way based on the join condition. And if you create a graph like this so I I I would know that in which order I need to look up so that I can complete my join. And the next part was uh this time we did not want to keep all the data on TDB. So we wanted an index on lake an index that will tell us uh uh the data about an ID faster than before and let's see how that secondary index works. So secondary index is a mini table. It is created uh as a subset of the original table. It has few columns. It contains your primary key and couple of join columns and then your partition value. So in in our case all our data was partitioned on date. So partition value was mostly date for us. And if you have a mapping like this where it says for this ID this is the date. So first you can reach the index and get the date and then you can actually go to the original source table for that partition and get the actual data. So if you look at the query on the left side uh right side the uh there is a nested query right it is picking all the created date from your index then you are passing that as a list of values in your main query. So that is how you do the lookup faster using the secondary index. So we can use the same secondary index to query on ID and query on the join column. So if you look at the second query here I'm looking up on payments where order ID equals to this. So this solved the lookups for us. Now the next part was how do we propagate the change events to the fat. So let's say uh you got all the change events on payment from the updated date partition. Now you have to complete this denormalized row. So basically you have to do the enrichment. In order to do the enrichment you have to get details of cards, orders, transfers and settlements. But there is a twist here that if you want to get the details about settlements first you need to look up for transfers and that is where this graph relationship helps us that it will tell us in which order we have to look up and how do we join those rows. So once we have this graph created from the fact configuration we traverse it and do a level order traversal uh and then we use the secondary index to do the lookups. So I get the card details, order details using lookups and once I have all the details required for enrichment I join it and create a big denormalized data frame which I can upsert on the target table. So now if you look at most of the databases like click house shift they give you an option of materialized view right so which you can create for premputing these joints but there is a problem there that it supports only changes on the primary table. If something changes on the primary table, it will propagate. But if something changes on the secondary table, it will not go in. So I will tell you why you need that. So consider a case of a payment that happened today. Now the amount is with razor pay nodal account and we have to settle that amount back to the merchant. So this can easily take a couple of days to weeks. So when settlement happens, we get a UTR number from the bank which we need to inform the merchant that so this is the UTR number where your payments were settled to your account. Now when the settlement happens you don't get a change event on payment. So payment has not changed only the settlement is the new event that you have got. Since we have to give the accurate data in the reports, we have to make sure that settlement UTR also reach uses your denormalized table and [snorts] uh graph actually helped us there because we uh from the graph we can know that in order to get the payment rows I need to do a backwards. So from the settlement I do the lookups on transfers and from transfers I do the lookup on payment and now I know which are all payments are related to my settlement. So once I know the payment rows I can go to my target fat and update all those payments. So this is how we solved the updates on secondary table. So now there there was one final piece of the puzzle that we did not want to keep dimension tables and high cardinality tables in the fact. So as I said before uh the right amplification problem with payment links. So we did not want to keep payment links as part of the fact because that would result in right amplification. So uh we wanted to do the joins at the runtime. So if you have to do a join of payment and a payment links factor, payments factor and payment links. So you can look at this query on the left side. So here you are doing the join and passing couple of filters. You're passing a merchant ID filter and a date filter. Right? Now Spark will pass this filter only on the primary table because you don't have any filters on the secondary table. So that that results in a full table scan on payment links and this is what we wanted to avoid during runtime joins and what we did is we did a bucketing on merchant ID so on both the sides and we added one more clause here saying that fact dot merchant ID equals to payment links do merchant ID now spark has this additional information that merchant ID is a common predicate between both and it can push that predicate to both the sources When we did this, it was able to push down the predicate on the payment links as well and it reduced the scan for us. And these are the numbers from that. So we had uh around 2TB of data in payment links without bucketing it was doing a scan of 2TB and when we added the bucketing it reduced to 180GB and within the bucket we also did a sorting on merchant ID. So when you do the sorting on merchant ID you will get a narrow uh metadata for a file. So you will have a narrow min and max so that you will be reading less number of files compared to earlier. So this reduced the scans and we chose iceberg for most of our new tables. Uh iceberg gave us the bucketing and sorting compaction option and also the major reason is the metadata pruning. The way it handles metadata pruning is far better than the earlier table formats that we used. And this is how the uh architecture evolved. We had a single data pipeline that was doing the denormalization and creating a fat that got changed to three different pipelines. One that generates your secondary index and another that generates your replication tables party bucketed on merchant ID and then you have a incremental incremental batch pipeline which will actually do your uh incremental processing of fat and these are the numbers that we got. We migrated five of the fats to the new strategy and we were able to cut down compute from 30 plus hours to less than 3 hours. This was straight 90% reduction and we got the same reduction on the cost as well and a large fat actually reduced from 8 hours to 1 hour. We were able to cut down on the tidb scan window and uh now we were able to support the full history on the fat. So once we do a back fill with full refresh and then we can incrementally update that fat. So you can maintain the full history in the denormalized and along the journey we learned lot of lessons and these are the few lessons I want to stress upon. So one is uh streaming stream processing did not work for us for doing the denormalization and if you have seen the approach that we took uh we were doing lot of uh lookups on tidb and our joints were mostly stateful joins so uh doing a one-time bad job per day was more cost efficient than running a stream job throughout the day and the next is we thought that lake would be mostly used for analytical purpose where we do some kind of aggregation but we never knew that there would be lookups coming into data lake and you need to optimize for that. So we created secondary index using that we were able to speed up the uh lookups on data lake. And the next is uh if you are modeling your fact based on the query patterns uh never create a big super fact which can serve all your queries. try to model the fact in a domain based modeling where you create fact for a particular domain and then uh you keep some of the tables which are less frequently query uh as a separate table. So you can do the joints at the runtime if you want to uh but do not create a very bulky fact with all the tables. And the last is so we were treating fact as one single white table and that's when we were doing the full refresh but if you actually look at how a fact is generated it is basically a group of entities right and these entities have a relationship. So if you know the relationship as in how those entities are related, you can recomputee only the affected data and not the complete data and that's how we did the incremental process. Yeah. So that's what I wanted to cover in this talk today and thank you so much for joining in. Yeah. Open to questions. First of all, thank you for the talk. It was quite interesting. Uh one question I had was like about how you kind of verify uh data quality uh especially when you're kind of going through these iterative uh fact recreation and all of that right so uh how do you ensure that uh you know the reporting service for example is looking at the accurate data against your OLTP and that sort of thing. See um for us moving from this full refresh to incremental right this was a new strategy that we introduced and we had to make sure that whatever the facts that we are generating it is accurate and it is matching with what the full refresh would do right so full refresh was the baseline for us and we were comparing it with the full refresh fact. So in the initial uh days what we did is we would generate a full refresh pack for a couple of high merch high high volume merchants and then we would compare that with the incremental fact and see if the data is matching and also we uh I think this is a very good question in terms of production readiness uh because this came up when we wanted to take this live and people asked this question that how do you ensure that different pipelines that you have they are producing the correct output. So we created DQ jobs for each of the pipeline. One for secondary index, one for your uh change events and another for your application tables. And for application tables, we use the actual source microser tables as a uh uh baseline. And then uh we would do a random sampling and not the complete data comparison because of the data size we have and that actually helped us to flag uh issues beforehand. Uh I think just looking at it from uh sorry just one more question. So uh if you were kind of like to take a fresh look at this right like uh how would you change it like in the sense that um like would you say that like uh splitting into microservices was fair at that time looking back now? And secondly uh like would you look at a different approach like you know maybe sharding uh and maybe horizontal partitioning to kind of uh alleviate some of these issues that you had. >> So uh bucketing is one of that approach where you did the sharding of the data right based on merchant ID. So most of the reports that we generate are for a particular merchant and if the if you know that all the related data for a merchant is sitting in one bucket you can do the join faster. So that that's we where we did this bucketing as a proxy for sharding and if you have to reook at the architecture. So one is we were using this tidb right I think now sunny was mentioning that they have now uh S3 backed uh storage right and keeping all the data on EBS was super expensive for us. I think with this option of S3 maybe we can explore u the joints on TIDB can be faster and we don't uh incur a lot of cost in storing that on EBS. Uh thanks Amed this is Vive here from Startry. So the question on um the SLA that you moved from uh multiple hours to 3 hours right is that accept acceptable SLA right now or are you still trying to improvise and reduce the time taking to generate a report? See for report uh the freshness gap is less than 5 minute the SLA is less than 5 minute and uh since facts cannot refresh that often right so we have 24 hours of freshness even for the incremental fact so this remaining gap that we have we serve it from the TIDB so TIDB uh the replication latency for TDB is within 5 minutes so uh whatever uh data gap that we have we read that from TIDB do a merge of uh incremental plus TDB to serve the reports so reports will never go stale But that is the data freshness part of it, right? But the report generation is still taking 3 hours today. >> No. Uh this is not the report generation. These are only the premputed tables. Report generation is still within 10 minutes. >> Got it. >> So once you have this denormalized fact, right? So you don't have to do the joins when you're generating a report. You are just doing a basic filtering saying that these are the dates and this is my merchant ID. You you're just doing basic filtering, no joints. So report generation would still be faster in few minutes. Got >> it. Thanks >> Amit. I have a question. Thank you for the presentation. Um wanted to know uh the uh reports which are generated for the users, right? Which are basically your partners, vendors etc. How much variance uh is there in the kind of reports that they uh they ask for and what impact does that have on your design? I think that was uh super critical for us because initially when we started we had maybe around 10 templates uh the settlements report transaction payments like that but as data size increased and we started shipping more and more features more products uh the number of templates increased to even around 60 70 these are standard templates I'm talking about which are available to all the merchants and then merchant can come to our uh account manager or the relationship manager and say that I want a certain data uh which is not available in your standard template Right? And he can create his own custom report template as well. And if you look at the overall number of templates, these are more than 3,000. Uh the custom templates are too many because merchants have a diverse set of requirements. And we tried to fit most of the reports as a based on a JSON based DSL where you define these are the tables you want to join and these are the filters. And running these kind of reports were easier because you have a DSL in front of it. But there are some customers the very enterprise customers like ATIL, IRCTC. So they have very custom requirements and for them we had to write a raw SQL uh plain SQL that actually does the join and these SQL are very uh expensive as in they have a lot of nesting and other things. Yeah. So we had a diverse set of templates to support and that actually created this much of mess because we had to uh have so many of premputed fact otherwise we would have gotten with one or two precomputed fact. >> Okay. Thank you. >> I had another question about you know analytical queries. So uh so so yeah so uh I had another question about analytical queries. So I'm assuming that you'll probably have like another team that is monitoring uh you know or maybe doing business intelligence on top of this and for that they'll need analytical kind of queries. So does this uh also support that workload or do you have another system for that? >> Uh so we uh use this as a common layer for both reporting and analytics. Uh but and also for analytics they don't want a most historical data as in a year back or two years back right. So for them we have used click house as one of the solution where we replicate some of the data to click house so that their dashboard and other things can run very fast. Uh because here we are doing still a lambda merge between IDB and fact. So for them it was not feasible because they have to show that in a dashboard and they want a real-time data and we use click house for most analytical queries. Ju just a quick question. You talked about this secondary index and you talked about the p the graph traversal for uh going up and down the graph, right? So question is how complex are these uh graphs that you have to deal with and uh which technique yields what kind of outcome? Would one of the two suffice for you or would did you have to do both or both were equally relevant for your solution? >> Yes. So yeah, so both are equally relevant. Uh so this graph is not actually a graph DB. It's just your dependency graph that you store it in your uh program that says this is how the entities are linked and if I have to recomputee the denormalized row, I can traverse that graph up and down and create that uh denormalized row. Right? And secondary index was important for us because uh we wanted to keep all our data on lake because lake provides that cheaper option that you have you can set it up on S3 and it is far cheaper than compared to TDB or other solutions. So if you have to do the lookups on lake uh the only option we needed was a kind of an index because lake does not have a native index capability. So we built the secondary index so that we can do the lookups faster and graph helped us to know which are all the related data we need to recomputee instead of full report. Yes. Correct. Correct.