Using a database in a data science project - Lessons learned in production

Speaker 1 [00:04]

This is my name is Jacopo and I work in Flixbus in Berlin science now four years and today I'm going to talk to you about some lessons we learned about using Postgres or in general any relational database in a data science project that is in production so by production I mean that we have to process a lot of data to give you an idea in peak periods so not during covet but now and before we have up to five tickets sold per second we have to process this data at least every day and it must be fresh so we have to be sure that we do not see all data otherwise our results are wrong what are we trying to do essentially what my team does is we get data from the past from the past years and we see how many people did buy a ticket for a given ride between a a specific from and to stop, and how much did they pay, and when did they buy the ticket. Given that, we build a model, we train a model, and we use it to predict how many people would like to buy a ticket for every ride in the future, in general in one year in the future, for every specific stop. So as you can imagine, it's a lot of data, it's millions of predictions, and we recalculate it every day, with different time periods, different models, but essentially it's a lot of data. So we have the problem of performance, of course, because it's a lot of data and can take a lot of time to process. And we have the problem of stability, because as we keep adding more lines, merging with more companies, changing the services, or just changing the infrastructure, we have the schema that keeps changing, so we need to deal with that. And we have to be very careful about outages, because as you can imagine, this kind of prediction, so the prediction of the demand, is what we use to then calculate the price, which means that if we don't calculate it one day, we lose money pretty much. So it has to be very robust. So today we are going to focus on the database part of it. So not how we do the machine learning, but specifically how we read and write data on the database. We have lots of data, so performance issues may arise. There are many Python libraries that you can use. You have many queries and you need to keep track of how much time it would take. So is a query stuck? Are they actually running? Have they taken too much time? And we need to deal with schema changes. So you have new tables, new columns, or data that is in a different format. This is going to be very specific for our use case, which means we use a lot of historical data. If you don't, then I probably don't really care. And we have mostly batch operations, like training and prediction, are generally running in the ground. so you don't really need a specific time. You just care about the overall timing. But we also have some web apps. So we let users, internal users, alter the forecast. So they can specify that the forecast for some day is higher or lower than we predicted, because there is maybe a concert, a strike, some event that we just cannot integrate in our model. So humans can actually change it. And that means that we have to provide web apps for that. And this is important, because if we use, for example, a columnar data storage, like Redshift, that's a bit tricky to do. It's possible, but ugly. So we need also some sort of punctual access and operational access to our system. We use Python because we are at PyCon, otherwise we would be at another conference, and it's because it's essentially the easiest way to do data science at the moment, and Postgres because it's basically, that's my opinion, but it's based on facts, it's basically the best database that you can have for free. It has a lot of functionality, it's very performant, and it's free, open source, it's nice. This is how usually companies do this. You have an operational system, so the system that actually does your business. In our case it's the shop where people buy the ticket. It can be an app or a website and an app or whatever, But basically this is a system that has a database which is designed for operational usage, which means that you access always, you have usually a lot of concurrent connections, and each one reads or writes a very small amount of database, usually single rows. So you have a sort of rain of continuous small operations. And this is an operational system. You care about very fresh data, of course, but you don't care about historical data. Then you have an ETL, so some operation that runs usually overnight, which reads from the operational system and writes the data into some database, like Redshift or BigQuery or Vertica, if somebody uses that, that is able to allow analytical queries in a very fast way. So you can easily perform something like a group by your account on a lot of data fast, It's not that good for operational access, but that's fine. Also you can do some transformations. So you can do joins, group I, filters, and get data in a format that's nicer for analysis and reports and also potentially machine learning. We don't do that for a few reasons. The first is that we need fresh data. So we cannot accept a nightly dump. We did in the past, but now not anymore. We need data that is pretty much real time. And to do that, we use Kafka. So we have a tool called Dbizium, which, in case you don't know, essentially intercepts every operation that happens on a MySQL database and detects every change that happens on a database. And based on that, it generates an event on Kafka. So every time somebody writes or updates a row on a database, you get a Kafka event, which is very simple to parse. So what we do, we read from these Kafka events, and in real time, essentially, we write the same change on our database, which is a Postgres database. This means that we end up with an exact copy of the monolith database, of the operational database. The problem with doing that is that we essentially cannot do any ETL. We have an exact copy. So we can choose the tables that we want, that we can do. We can choose the columns. We can do some stuff like changing the time zone, but that's it. We cannot do joins because we have independent tables that we don't receive in the same order in which we are created so we cannot possibly do any operation on that we can only see an extra copy of the production database so we do a very simple trigger thing but very powerful which is we use materialized views so in a podcast but most relational databases you can find a view which is just a query with an alias so when you need to query the same data again again you can find a view and say okay this long big query with joins and group bias and so on as an alias and you then query that alias instead of the actual query which means it's nice for the developers you don't have to see the complexity so it's like a function in python you just call it and don't care about the content but it doesn't do anything for the performance if the database is simply replacing your alias with the actual view every time you run it so it's not faster than just running a bigger query it's just nicer to see With a materialized view, you essentially ask the database to take a snapshot of what you will see through a query, and store it as a physical table. So you can then apply indexes on the table, and accessing it is as fast as accessing any table. But of course it's a snapshot, so you have to refresh it manually. So we just use it. This is nothing strange, because you can do, many companies they do that, and we can have a cron job on Kubernetes or just any operation to refresh it. The thing that we learned is that if we keep a naming convention so we can associate any materialized view with a corresponding view, we can then easily switch between views and materialized views and see the performance between the two usages, and essentially it's much easier to sacrifice when needed, speed for freshness, so we can always decide whether to use the view or the materialized view, and it's very easy to compare them. Then, okay, we have a problem. The problem is that if we have applications that do a lot of queries and wait a lot of time performing queries, we have to monitor them, and a problem that everyone has at some point is the speed of the application has to be improved, of course. And when that happens, what many people do, and that's a big mistake that every developer does, is instead of profiling the application and knowing how the application spends time, they just imagine something that they decided is the best way to optimise. Think about tail recursion. At some point, it was like the hype and everyone was doing tail recursion. Now, instead, we want to know how the application is actually spending time. For example, how much does it take to run a query on average? You can always put the query in a database client, run it once, but that's not indicative of how much it takes to run it on actual loads. So our solution is actually multiple steps. The first step is to have a SQL file for each query. This is simple and is a very easy and effective way to keep your query in order. So you don't have strings, parsed around all your application. You don't have even worse functions that generate pieces of query that then you compose because then you have no idea what you're running. You have a folder with SQL files, and every SQL file is a query. This makes a lot easier already to monitor them. You can easily see them. You can edit them. You can do static analysis. So there are tools that can parse a query and tell you which tables and columns are being used. So you can do a lot of nice stuff automatically. But then it enables a few nice things that we do. The first is that you can define an helper, which is how we actually do queries. So whenever we want to run a query, we simply call this helper and say, okay, run this query, call it in this case get relational popularity. The name will be get relational popularity dot SQL. We just omit it in the helper. And we pass the parameters. So this is already nice because you don't have much code. You just invoke the query directly. But you can do a few fancy things. The first is that you can retry. So if you get an error like a network error or database is down, you just wait a few seconds, retry, and then you have a sort of exponential backoff or whatever logic you want to do. And this is how we can actually upgrade the database. So we have just a Postgres instance on RDS. We sometimes upgrade it. We pass from Postgres 13 to Postgres 14 and so on. This has a small downtime. Thanks to this helper and retry logic that is in it, our application keeps running. It just waits and then keeps running. And essentially, we never have to worry about that when we run training or prediction. The second is that we can have some transforming. These are just nice things to have. So if we get a Pandas data frame, and since we have data frames are easily gigabyte of data big, we want to get the X at, how to say, the number have to be in unsigned integer or integer, but get specific format because that means the difference, that makes the difference between crashing or not because we consume all the memory. And then, and this is actually the point here, we can do profiling. ScienceVis helper is the only place when we run a query, we can always, whenever we run a query, see when it starts, when it ends, and how many times it was executed. So, we can easily generate something like this. You don't have to read all the lines, because that's very specific. The important thing is the concept. When you run a program, you want to build a profile. You want to see how actually it is spending time. And by doing that, by having a simple helper that invokes every query that runs them, we can easily collect statistics, and And not on just one execution manually, but every single execution in production. And then we easily build such a thing. So a profile that shows you how the program is spending its time across all the queries and potentially every Python operation. And it's very, it's quite normal that you have a sort of Pareto distribution of the time. So your program spends 80% of the time doing 20% of the queries. So it's better to have this information before optimizing. Another advantage is that you can detect in advance when a query is taking more time because if you have a lock or a table that is growing, a query can become slower and slower, and you want to detect it before it's so slow that your program crashes. You want to see it in advance, so you can use the Datadog or New Relic or whatever and monitor this data in a systematic way. This was all about reading data, which is fine, but then you need also to write data because if you calculate predictions, you have to write them somewhere. We write even more Kafka usually, but also on our database. So how do we do millions of rows on insertions every day in a decent, fast way? There are many options. Assuming you use SQLAlchemy, execute many is probably bare minimum. It opens a transaction, does a lot of inserts and closes the transaction, which means it's faster than doing one-row per transaction, of course, but still you can do better. The best way by far in Postgres is the copy command. Copy essentially gets a CSV of the data and uploads it directly to a table. It's a lot faster than every other option. The problem is that it's very ugly if you have JSON or data types that are very complex to represent. Imagine, for example, PostGIS and geometries, and also the interface is quite ugly. With Psychobg2, you have to write QuartZone code, and it's not very user-friendly in case of errors. And in particular, you cannot deal with upsets. So if you insert data, you want to deal with data that is already there, and you want to specify a logic to merge this data, and copy cannot do that directly. What you can do instead, you can use a logger tables, and a logger table is a Postgres functionality that essentially creates a table which is sacrificing, how to say, persistence. So you lose the table if the database crashes. So it's not what you want, usually. But it is a lot faster to read and write. It can be easily two, three times faster when you write it on an Unlocker table. So you can copy inside the Unlocker table, eventually just merge it with the final data that you have. in PsychoPG 2, you also have this very ugly functionality called execute values, which essentially expects a template for an update or insert, and it unrolls the template, replacing a sort of percentage S that you write yourself with the actual data. So basically it builds a huge query by repeating values in the query. The problem is that it's a bit ugly to see and is impossible to troubleshoot. So basically when you have an error, you have no idea which columns caused it. You just have to try until it works. But in my measurements, of course that depends on your data, it was around 20 times faster than just normal insertions. And then my favorite personal option is prepare statements. Prepare statements allows you to send a query to a database which can be very complex. The database compiles it and in the recent version of Postgres as this happens through LLVM, so it's very fast. And then you just call the handler for the compiler query by passing parameters again and again. So basically you can do a lot of insertion, calling the same query again and again, in a very effective way. PsychoPG3 has it as a sort of first class function, it's already there, it's nice. And also PsychoPG3 supports the binary protocol, so it can communicate with database, not using text, but sort of byte representation it is much more efficient and compact. The problem is that PsychoPG3 is not supported yet by SQLAlchemy. SQLAlchemy 2 will support it, but the current SQLAlchemy doesn't. And SQLAlchemy 2 is still not available as a stable. I think not even beta. So that is the future, probably, but it's not there. You can do PsychoPG2, but it's ugly as well. And here I wrote an article about comparison of these methods and many more. By the way, I published a slide on Discord, so you can access them. Another trick that we found is that we can simply pre-process data once and store it as file. So when you do training, you reread the same data again and again over time. So it's a bit stupid to redo the same query, to re-get the same data that is in the past. So it's not supposed to change. The ride already departed, the tickets already bought. So it's not different today from tomorrow. If it's a ride that was one year ago. So what we do instead, we calculate this data day by day for the past and we get the result which is a huge aggregation of everything that we may need as a file and we store it on S3 or any data storage. By doing that you have a huge performance improvement. I actually was very surprised when I saw how fast this can be. We ran a comparison which could be a talk by itself, but basically we ran a comparison of every format that we could find and we found out that Parquet and Arrow is probably best combination. It's very fast, compressed a lot, and the support on Pandas and Spark is amazing. So, essentially, Parquet is a layout that allows for columnar access. So, when you create a file or multiple files representing some data, you can later decide which columns you want, and you just read them. So, when you read, you can save a lot of time by reading only what you need. Arrow is a It's a bit hard to define, but you can imagine it as a sort of pickle across languages. So you just store your data using Arrow, and you can read it back without thinking, oh, yeah, I'm using NumPy 120, and I cannot read it because it will pick up another file. No, nothing like that. You just read and write from Java to Python to whatever. It does actually a lot more because it's also a way to represent data in memory, but we don't care about that. And then we compress with snappy, but that's just not the difference from BZO2. And most importantly, with Pandas, it's super trivial to create and to read and write a file in this format. You just install PyArrow, and you have a method like save as arrow, I don't remember the name, but basically store in this format, and then you read it back. Then, as anticipated, we have the problem of handling the schema changes. These can be done in many ways. One of the most common is to use SQLAlchemy because it has a migration functionality. SQLAlchemy is an ORM, so it assumes that you basically want to use the ORM functionality together with the migration, which we don't use. So it's nice as long as you have a format that plays well with the ORM. It's not nice when you have, for example, user roles, and you have grants that we do, views, you have some complex data types. In that case, you have to write them manually, and you basically defeat the purpose of having SQL Acme. Then you have Skitch, which is on the other end of the spectrum. Skitch allows you to define all the schema through SQL that you write manually, and you You can also write functions to check what is the state of the database, so you can just ask it to check the state of the database and apply what's needed. So it's extremely powerful, but you have to write a lot of stuff by hand, which for us is overkill. So what we do instead is a very primitive approach. We have a Git repository with a schema.sql file, and that's it. So it's a schema, you get an empty database, you run it, and you get all the structures that you need. Then you have, of course, the problem of finding out what is the difference with the database. We have this monstrosity that is actually much more simple than it looks, is our real makefile, and we use two simple tricks. The first is that there is a tool called pgVirtualEnv, which can create a sort of on-demand instance, Postgres instance, and so by doing that, we can create an empty instance and apply our schema to that instance then we use pgdump which is just the postgres uh default way to dump a database with the option schema only which only gives you the schema so by applying pgdump schema only to our ephemeral database created by pg virtual and to the remote database that is what we actually have we get two schemas schematas depending on what you want to use and we can just do a textual diff. It's just text. We have a huge blob of SQL. We just compare them, and we can easily, very easily find out the differences between the schema that is in production or staging and the desired schema. So we avoid having tables that are completely wide and nobody knows that they even exist. The final trick, I mean, the final for what we can fit in a talk, is to use PostgreSQL table. So PostgreSQL has a lot of nice tables that tells you everything about what is happening in the database. So how much an index is used, how much a table has sequential accesses, so accesses that are slow because there is no index being used, logs, running queries, and so on. And I can suggest this amazing website, I didn't do it, I just found it, which is pgstats.dev, which has this chart of all the tables across versions and what they do and which kind of data they show so it's really an amazing resource and the more you learn to use it the more black magic you can do with your podcast instance and finally we are adding tends to call it now everyone has a qr reader on their phone and so if you are interested in these kind of technologies and apply to a actual real problem in which we have a lot of data, real data, that we apply to a business-relevant problem, you can apply and ask us. We also have a stand on the third floor, so see. And so you can just reach out to us. If you have questions, this is a good time to ask.

Speaker 2 [23:18]

Thank you for the talk. It was very interesting. I wonder if someone has questions. If you have questions, you raise your hand and I will run to you. And yes, I don't see questions and it's little. But yeah, we're going to take offline questions now.

Speaker 3 [23:46]

First, thank you for this interesting talk, but I missed the part when you were talking about CDC and the stream of event which is coming from Kafka, and then you jumped to using materialized view on Postgres. Is it like you started to store the events in Postgres and then to do the joins from multiple streams?

Speaker 1 [24:06]

streams yes so there is this usually you have an etl and the etl doesn't just copy data as is but also rearrange it in a way that is nicer for you in our case for example we have distances which are a bit tricky to calculate and because a distance can take multiple countries and have different versions but we don't care about it we just want the distance so with an etl you can have this calculation done for you but if we use kafka we get fresh data we have a database in real time but we lose that. We cannot do any ETL, we just have exactly what we have in the production database. So we need some way to perform some sort of mini ETL on the fly, and that's why we use materialized view. So basically materialized view replaces the operation that an ETL would normally do for you.

Speaker 3 [24:53]

Okay, so you are storing the events in Postgres and then doing the joins in Postgres but using better materialized view to improve the performance of the joins.

Speaker 1 [25:01]

Yes, so we don't store the events as they are. Since every event represents an insert, update, or delete operation, we redo the same operation on our database. So we have an ID, and when we see that something was updated on the monolith database, on the operational database, we apply the same update on our database. Deletion and insertion are the same. Then my question.

Speaker 3 [25:23]

Then my question will be...

Speaker 1 [25:24]

Oh

Speaker 3 [25:25]

OK, last question. OK, last question. So is it possible to do this using stream processing, like not to store it in Postgres and then to do the joins? This is my question. This is my .

Speaker 1 [25:37]

No, you can do some string processing in Kafka, but by the nature of Kafka, you can always do with Windows. You cannot do with all the data. And on Postgres, it's the opposite. You always see the whole data, and you have to basically refresh the material view every time. You can use partitioning, do some magic, but that's essentially it.

Speaker 2 [25:56]

Thank you for the question. Does someone else have a question? I want to give opportunity to his.

Speaker 1 [26:03]

How do you handle locks with the migrations and stuff like that? You say locks? Yeah. We don't. Well, essentially, when we migrate, we just know what we are doing, because that's the database I work with every day. If I have locks, either I wait, because I know that there is a prediction job that is running and I don't want to touch it. Otherwise, there is a function which is pgterminate pit, so you can just ask the database to kill every connection. And you can just terminate everything and there are no logs. One question. So materialised views, they are supposed to be refreshed every time you get new data? No. That depends on the application. But usually, you either do a cron job, so every day at some point you just update it, or every week, depending on your case. Or Or if you want, you can basically keep track of what changes in the materialized view. So when there is an update, depending on how often the data changes, and materialize only on demand. That's another thing you can do. That depends on your application. So you basically decide how much to sacrifice speed to access to get fresh data. But aren't materialized views constrained to only materializing the whole data set? Wouldn't that be a bit... They just materialise a query. So it's like running a query and getting the result into a table. You don't have to apply them on a view. So actually you can apply them to a query that you run on the fly, but in our case we just prefer to have a view and a materialised view based on that, like in that case. If you want to get all the recent data, you just put a where in the query and you just get whatever it comes out of that.

Speaker 2 [27:58]

Does anyone else have questions? So otherwise, I thank you. A big applause again for the talk.