Building reliable data processing - lambda architecture inside Google BigQuery
In this article I want to share a way that allowed us to end the chaos with data processing. I used to consider this chaos and subsequent re-processing to be inevitable, but now we have forgotten what it is. I give an example implementation on BiqQuery, but the trick is quite universal.
We have a very standard process for working with data. The raw data in the most raw form is regularly uploaded to a single repository, in our case, BigQuery. From some sources (our own production), data comes every hour, from others (usually third-party sources) the data comes daily.
Subsequently, the data is processed to a usable state by a variety of users. These may be internal dashboards; reports to partners; results that go into production and affect the behavior of the product. These operations can be quite complex and include multiple data sources. But for the most part, we deal with this inside BigQuery using SQL + UDF. The results are stored in separate tables in the same place.
The obvious way to organize this processing is to create a schedule of operations. If the data is loaded daily at one in the morning, then we will set the processing to 01:05. If this data source is loaded in the region of the 5th minute of each hour, then we will configure processing for the 10th minute of each hour. Gaps of 5 minutes are not critical for users and it is assumed that everything should work.
But the world is cruel! Data does not always arrive on time. Or they don’t come at all, if not repaired. If your hourly download ended at the 11th minute, and the transformation started at the 10th minute, then please wait another hour to see this data in the dashboard. And if the operation uses several sources, then the situation will be even more fun.
Moreover, the loaded raw data is not always true (the data is generally always incorrect!). Periodically, the data must be cleaned or reloaded. And then you need to restart all operations with the correct parameters so that everything is repaired.
This is all, of course, problems with raw data and it is necessary to solve them. But this is a war that cannot be completely defeated. Something will still be broken. If the data source is internal, then your developers will be busy with new cool features, and not with the reliability of tracking. If this is third-party data, then generally a pipe. I wish that at least the processing did not get in the way and as soon as the raw data was repaired, all customers immediately saw the correct results.
This is a really big problem. And how else to decide?
If processing leads to problems, then don’t do it! No need to do any processing at all and store intermediate results. As soon as the user needs the results, everything should be calculated on the fly from the raw data. Given the speed of BigQuery, this is quite realistic. Especially if all you do with the data is GROUP BY date and count (1), and you only need data for the last 14 days.
Most analytics work specifically with such queries. Therefore, we are actively using this solution. But this approach does not work with complex transformations.
One problem is code complexity. If you add all the operations into a single SQL query, then it will not be read. Fortunately, this is solved by view tables.(representation). These are logical tables in BigQuery, data is not stored in them, but is generated from an SQL query on the fly. This greatly simplifies the code.
But another problem is performance. Everything is bad here. It doesn’t matter what fast and cheap modern databases. If you run a complex transformation on one year of historical data, it will take time and cost money. There are no other options. This problem makes this strategy inapplicable in a rather large percentage of cases.
If there is no way to do without a processing control system, then you need to build this system well. Not just a cron script execution schedule, but a data loading monitoring system that determines when and what transformations to start. Probably the pub / sub pattern is very suitable here.
But there is a problem. If building a complex system is more or less simple, then maintaining it and catching bugs is very difficult. The more code, the more problems.
Fortunately, there is a third solution.
Lambda architecture is a famous approach to data processing that takes advantage of scheduled and real-time data processing:
* I don’t know how to properly translate into Russian, is a batch job a batch job? Who knows, tell me!
Usually this is all built using multiple solutions. But we use essentially the same trick just inside BigQuery.
And here's how it works:
Scheduled Processing (Batch layer). Every day we execute SQL queries that transform the data currently available, and save the results in tables. All queries have the following structure:
The results of this query will be stored in table_static (overwrite it). Yes, BigQuery allows you to save query results in the table that was used in this query. As a result, we take the old, already calculated data (so as not to recount it) and combine it with the new data. X days is the selected period for which we want to recalculate the data in order to take into account all possible adjustments to the raw data. It is assumed that in X days (how much is individually for the source), all adjustments will already be made, everything that is broken will be repaired and the data will no longer change.
Real-time access (Speed layer + Serving layer). Both of these tasks are combined into a single SQL query:
Yes, this is the same request! We save it as a view with the name table_live and all users (dashboards, other queries, etc.) draw the results from this view. Since views in BigQuery are stored at a logical level (only a query, not data), each time it is accessed, it will recalculate the last X days on the fly and all changes in the original data will be reflected in the results.
Since the request is the same in both cases, in reality, to avoid code duplication, the daily request (from the batch layer) looks like this:
This approach has several important advantages:
PS If you like to use tables broken by dates in BigQuery (we love it very much), then there is a solution for this. But this is a topic for another post. Hint - the functions for working with these tables do not swear, if some of the tables are just representations.
PPS If views in BigQuery supported caching (how it works with regular queries), that would be really cool. This would essentially make them materialized views. And the effectiveness of our approach would become even higher. If you agree - here you can put an asterisk to implement this feature faster.
We have a very standard process for working with data. The raw data in the most raw form is regularly uploaded to a single repository, in our case, BigQuery. From some sources (our own production), data comes every hour, from others (usually third-party sources) the data comes daily.
Subsequently, the data is processed to a usable state by a variety of users. These may be internal dashboards; reports to partners; results that go into production and affect the behavior of the product. These operations can be quite complex and include multiple data sources. But for the most part, we deal with this inside BigQuery using SQL + UDF. The results are stored in separate tables in the same place.
The obvious way to organize this processing is to create a schedule of operations. If the data is loaded daily at one in the morning, then we will set the processing to 01:05. If this data source is loaded in the region of the 5th minute of each hour, then we will configure processing for the 10th minute of each hour. Gaps of 5 minutes are not critical for users and it is assumed that everything should work.
But the world is cruel! Data does not always arrive on time. Or they don’t come at all, if not repaired. If your hourly download ended at the 11th minute, and the transformation started at the 10th minute, then please wait another hour to see this data in the dashboard. And if the operation uses several sources, then the situation will be even more fun.
Moreover, the loaded raw data is not always true (the data is generally always incorrect!). Periodically, the data must be cleaned or reloaded. And then you need to restart all operations with the correct parameters so that everything is repaired.
This is all, of course, problems with raw data and it is necessary to solve them. But this is a war that cannot be completely defeated. Something will still be broken. If the data source is internal, then your developers will be busy with new cool features, and not with the reliability of tracking. If this is third-party data, then generally a pipe. I wish that at least the processing did not get in the way and as soon as the raw data was repaired, all customers immediately saw the correct results.
This is a really big problem. And how else to decide?
Solution # 1 - Remove Problem Details
If processing leads to problems, then don’t do it! No need to do any processing at all and store intermediate results. As soon as the user needs the results, everything should be calculated on the fly from the raw data. Given the speed of BigQuery, this is quite realistic. Especially if all you do with the data is GROUP BY date and count (1), and you only need data for the last 14 days.
Most analytics work specifically with such queries. Therefore, we are actively using this solution. But this approach does not work with complex transformations.
One problem is code complexity. If you add all the operations into a single SQL query, then it will not be read. Fortunately, this is solved by view tables.(representation). These are logical tables in BigQuery, data is not stored in them, but is generated from an SQL query on the fly. This greatly simplifies the code.
But another problem is performance. Everything is bad here. It doesn’t matter what fast and cheap modern databases. If you run a complex transformation on one year of historical data, it will take time and cost money. There are no other options. This problem makes this strategy inapplicable in a rather large percentage of cases.
Decision number 2 - build a complex system
If there is no way to do without a processing control system, then you need to build this system well. Not just a cron script execution schedule, but a data loading monitoring system that determines when and what transformations to start. Probably the pub / sub pattern is very suitable here.
But there is a problem. If building a complex system is more or less simple, then maintaining it and catching bugs is very difficult. The more code, the more problems.
Fortunately, there is a third solution.
Solution number 3 - lambda architecture! …well something like that
Lambda architecture is a famous approach to data processing that takes advantage of scheduled and real-time data processing:
* I don’t know how to properly translate into Russian, is a batch job a batch job? Who knows, tell me!
Usually this is all built using multiple solutions. But we use essentially the same trick just inside BigQuery.
And here's how it works:
Scheduled Processing (Batch layer). Every day we execute SQL queries that transform the data currently available, and save the results in tables. All queries have the following structure:
The results of this query will be stored in table_static (overwrite it). Yes, BigQuery allows you to save query results in the table that was used in this query. As a result, we take the old, already calculated data (so as not to recount it) and combine it with the new data. X days is the selected period for which we want to recalculate the data in order to take into account all possible adjustments to the raw data. It is assumed that in X days (how much is individually for the source), all adjustments will already be made, everything that is broken will be repaired and the data will no longer change.
Real-time access (Speed layer + Serving layer). Both of these tasks are combined into a single SQL query:
Yes, this is the same request! We save it as a view with the name table_live and all users (dashboards, other queries, etc.) draw the results from this view. Since views in BigQuery are stored at a logical level (only a query, not data), each time it is accessed, it will recalculate the last X days on the fly and all changes in the original data will be reflected in the results.
Since the request is the same in both cases, in reality, to avoid code duplication, the daily request (from the batch layer) looks like this:
SELECT * FROM table_live
This approach has several important advantages:
- Each user receives relevant results, there is no resynchronization between raw data and aggregated data (this is provided that all jambs with raw data are allowed for X days)
- Users get results quickly. No one recounts 2 years of data each time they are accessed. Recalculation of the last X days in BigQuery is acceptably fast. X is chosen so as not to create performance problems. Of course, you can switch to hours instead of days, of course, but you did not have to do this with our data.
- No need to think about the schedule of daily operations. You just need to do this once a day, but it does not depend on the time of loading the raw data. In addition, there are no problems if the transformation falls in one day or if it is launched twice. For the user to notice the problem, it is necessary that the processing "lay" for more than X days.
- To build such a system, you need a minimal amount of code (read - problems). The SQL query itself is 10 lines longer than the original and is now stored inside BigQuery as a view. Plus you need to run daily processing, but this is an elementary task.
- The costs of this system can be even less than when creating a complex system of schedules. It seems like it should be more expensive since we are constantly recounting X days. But our experience shows that the opposite is true. If you upload data every hour, then you have to do a recount every hour, including weekends. And still have to recount X days ago to reflect the adjustments. Total for the week is typed 24 * 7 = 168 times. But in reality, the user can open this dashboard only three times a week. With our approach, we will have to count 7 times on a daily schedule and plus 3 times on the fly. Significantly less.
PS If you like to use tables broken by dates in BigQuery (we love it very much), then there is a solution for this. But this is a topic for another post. Hint - the functions for working with these tables do not swear, if some of the tables are just representations.
PPS If views in BigQuery supported caching (how it works with regular queries), that would be really cool. This would essentially make them materialized views. And the effectiveness of our approach would become even higher. If you agree - here you can put an asterisk to implement this feature faster.
