SBTB 2023: Jim Dowling, Bridging Python and Lakehouse worlds with Arrow Flight and DuckDB.
so my name is Jim I'm going to talk about how we can access data in The Lakehouse from Python and using two technologies one called Arrow flight and the other called duct DB and um if you don't know anything about lakehouses lake houses are a new type of data warehouse where the data is stored in a data lake so they're low cost and they give you much of the functionality you'll find in a data warehouse so you get asset updates time travel and great stuff at low cost so it's a very interesting place to to store a lot of data there are some competing table formats out there um there's been talked today about patchy Iceberg but we also have a Pache hoodie and Delta Lake ultimately they all store their data in either parket files or orc files um it's tabular data but what they've done is they've been designed really for the Big Data world so you'll see that you know there's clients for writing the data in sparker clients for writing the data in Flink and then there's query engines for reading the data so you can have trino or Dro and Snowflake and others in big query can can read the data in iceberg for example but what happens to python we have a lot of python developers out there a lot of them like to use pandas more using polers some could even use duck DB um how can we get at the data the increasing volumes of data that's being stored in the Lakehouse table formats from python so that's what I'm going to talk about today I'm going to talk specifically about hoodie where we've been doing work but I think a lot of it um applies also to Iceberg and Delta Lake and I'm going to start by talking about why we did this the reason we do it is because we're interested in building machine learning systems and particularly with a feature store which I'll get into in a second um but machine Learning Systems you can see as being effectively machine learning pipelines that are chained together and I don't tend to use the word machine machine learning pipeline because it's not well defined what the input and output is so we normally talk about a feature pipeline that takes raw data outputs features and labels training pipelines take features and labels and output a model and inference pipelines take a model and new features and they'll output predictions whether it's online or batch inference so given that this is a kind of typical machine Learning System the challenge we had was that in our feature store hops works we store the offline data the historical feature data in Apache hoodie we're also looking at Iceberg but at the moment we support a patchy hoodie and um you know we can write the data in the feature Pipeline and you can write it in Python you can write it in spark you can write it in Flink but when you write it in Python currently what we do is we write it to an intermediate layer to CFA in fact and then we run a spark job to write to the hoodie tables so we're using something called Delta stream for that and that's not exactly ideal and then when you're reading data into python your training pipeline is nearly always written in Python and if the data you're going to train your model on is in a table format like hoodie or or Iceberg how do you get the data out with high performance and the same is true for inference pipelines so online inference is nearly always in Python how do we get the data out efficiently so under the hood the feature store is actually a dual database system you have this offline store that I'm going to talk more about where we store all of the historical data typically in these table formats and then there's also a low latency store a row oriented store and for getting really low latency access to the latest feature data and the feature store is used to to basically manage your data for for machine learning for um you have the pipelines that feed it in and then we have training API for getting training data back uh batch API for getting batches of data frames back for batch inference and then also row oriented API for getting feature vectors for online models so we we've been working with this for a few years our feature store was the first open source feature store and we've provided python access via jdbc right so this kind of works like we have a serverless version of hsws it's got over 4,000 users on it and they're using jdbc to download they have been using jdbc to download the feature data stored in this table format into python client and it kind of works okay for small amounts of data but it's really slow um so we really needed to fix it and what we've done and what I'll talk about today is how we changed using jdbc to using uh Apache arrow and we also change the internal so we change from using spark and And Hive as our uh query engine to using duck DB so this is duct DB as many may know is a it's a it's an embedded Coler store it's this is an architecture I would say that's suitable for what we call reasonable scale data so if you're going to ingest data on the order of 100 gigabytes or if you're going to read data 100 gigabytes or less that's reasonable scale you can do it with these Technologies um okay so let's get get going on it um before we do the the the short note is we got a huge speed up so effectively what's happened is is using a feature store which all of the existing feature stores have this jdbc API they're very slow and it's not funny that feature stores don't have a great reputation amongst python developers because when you want to just get a gigabyte of data it's taking 20 minutes right and that's not reasonable um not for anybody who who can get a gigabyte of data from their laptop Mac Pro in in a few seconds so um what we see with with using arrow and uh duct DB is we're now getting Network line speed which is pretty uh incredible so let's get into it I'm going to start if you don't know what a pachy arrow is um it's an open source colner uh in-memory format it's an inmemory data format I should be say to be explicit so if you look in the uh chart here you can see that there's uh at the top of it we have rows Row one to four and you can see you've got these different uh columns that are colored blue green and red so that data is stored at the top in what we call row oriented uh format we have the uh the each row is stored continuously in memory and it's efficient when you want to retrieve a row of data so key Value Store can do this efficiently or even a a you know a database a relational database that allows you to do projections you can say just give me Row one in the First Column and if you um want to store that uh ently for analytics columnar formats are more efficient so if we pivot that data and you can see we have the arrow memory buffer at the bottom uh at the bottom corner of the screen you can see that we're now storing the data continuously in memor in memory by column value so we store all of the timestamp values continuously one after another in memory and this means it's much more efficient when you do analytics or Creer processing of the data in memory when you're doing a typical analytical operations so if you're Computing aggregation or doing uh complex joins and those type of operations so apart from the the memory layout we the colar memory layout we also have zeroc copy reads which is very useful and we'll exploit that today in our architecture what that basically means is that if we're transferring Arrow data from one process to another we may not need to deserialize and serialize the data so those of you who are familiar with the paty spark we'll know that there's been something called pandas udfs there for quite a while and pandas udfs are very effici ient uh way of of implementing um functions uh that take Panda's data frames as a a parameter and then they return a panda data frame or series and uh what they do is they transfer data uh from cor bark data frame to these workers that execute Panda's code so vectorize Panda's operation on the Arrow data and there's no copy involved so typically in spark world if you're running py spark there would be a copy of data would be deserialized from the jvm into the python code it if you run a python UDF it'll deserialize it send it back and it's all very slow so Arrow had uh been there for a few years and been very successful and but we'll see today that we use it also to transfer data from The Lakehouse all the way to Panda's clients and if you're familiar with pandas you may know that last year uh or earlier this year pandas 20 changed its internal memory format to Arrow so previously it was um uh numpy and now it's uh Arrow as its internal native format so there is another technology that we we're interested here which is called Arrow flight so aright is a it's an RPC protocol it's a way of transferring data from a client to a server from a server to a client so the client would say hey I want some Arrow data and it Returns the the the data in Arrow format so arrow is in fact also an over the network transport protocol format so you can send the data in Arrow it's check sumed and the client will validate the check sums and you can send the data straight over and if you compare that with let's say table you have and you want to send over jdbc in jdbc it's going to serialize the data and then deserialize it on the other side so what we can do U when we have a panda's client because if we go back and say well we want to build machine Learning Systems we're going to have these training pipelines implemented in uh in Python they're going to receive the data in pandas they're going to train their models and same with our inference pipelines they're going to be in Python want the data in pandas and what we wanted to do and we saw that well hang on arlight looks really really promising compared to jdbc because we can send Arrow data in an arrow table to a client without needing to deserialize or serialize the data and we also don't need to compar to jdbc do what we call a column to row pivot so if our data is in jdbc it's in row oriented format and when we want to uh use the data in pandas we want it in column oriented format in pandas so that means you might have to Pivot the data um from uh from jdbc to pandas so if we can avoid all that it's great and we looked at some of the the literature that's out there um Dro have done quite a good bit of work in the space on arof flight um and uh you can see some numbers here comparing uh jdbc and arrow flight the blue lower is better in this case it's the amount of time taken to transfer data columnar data from a client to a server and these were done by Dro these experiments but you can see it looks very promising compared to jdbc so even odbc is not great here there's not much difference because odbc is also a row oriented Network protocol so um if you have colmer data and you want to transfer it from a client to a server um Arrow flight is very interesting there is a new protocol called am adbc so Arrow database connectivity which is a layer on top of Arrow flight so it's basically like a jdbc driver that you would install and you could read from then a a server that's returning Arrow data to the client but we looked at arof flight as a kind of a more nrow option um because we want to just get back feature data for machine learning it's not an all a general purpose SQL API yet so arlight really promising um we had another issue that we saw internally with um with our feature store which was we using spark as the query engine and a feature store has data spread over many tables we call them feature groups and we have to join features together with our label in order to create training data sets at this join is quite complex it's called a point in time join it's actually kind of a a complex left as of a left outer as of join or as of left outer joint outer left join and um Spark's very slow at it you have to implement it yourself we looked at different techniques we made some optimizations compared to the spark query engine it's still not great but duck duck DB has a native implementation uh of that join which um is very highly performant so duck DB is an analy iCal database management system it's it's a Coler store kind of like datab bricks or snowflake but with no dependencies and you can install it inside your application as an embedded Library um you can install it in a juper notebook you can pip install duct TB and then you can send string you can read up data into it you can you know execute SQL queries on it and it's really fast it's it's competitive with even polers which is a very uh you know high performance data frame processing engine in in Python um you can put it inside ETL pipelines you can do what we did so what we did was we put it inside our server so we have an arlight server we have duck DB inside of it and duck DB is now going to do these point and time joins from the different feature groups to to create consistent snapshots of of training data for clients so um if you want to look at the numbers for duck DB it is it's really well engineered so the team who did this were the team that came from Monet so in Amsterdam they did the first Coler store Monet was the first real Coler store um so they have a lot of history in the space and if you look at the performance numbers in their latest Benchmark here this is a duct DB Benchmark but this is really for people who um you know are looking at kind of data at reasonable scale so you can see here spark is down at that we were looking at before 129 seconds um and duct DB is 8 seconds in this particular case it's a join Benchmark um with 100 million rows and seven columns so that's the kind of Benchmark we're interested in because we know that um you know that's one of the typical workloads we have in our feature store and you can see that it's really promising so that's great uh if you want to use duct DB in Python this is all you need to do you basically import duck DB obviously pip install ahead of time and then you can U have query strings and and execute them on the dctb instance so it's pretty straightforward and the interesting point that if you look at the last line here you can see that we can actually get the data back in Arrow format so duck DB can actually process data that's in Arrow format and we're going to exploit that later we'll see that we can read the parket data up from our table formats our Lakehouse formats into arrow and ddb can process that data in Arrow format and then we can return it to the clients the Panda's clients directly in Arrow format so this is the service we built uh we called it flying duck which was really cool we had a beard that had a flying duck in it and we were really nice with the branding and then duck DB said uh you're not allowed to call you're not allowed to use the term duck in any duck DB products and we're we're good actors so we we threw that in the bin uh so we're just calling it like arlight du DB or something arlight du DB I don't remember what we're calling it um but it's a query service for for creating feature data stored in the Lakehouse and it's powered by dub and AR flight all open source so excuse me so this is kind of what we had before just to put things in in uh context and this works right so you know you can use hops Works to kind of from a python client and it's great but the performance as I mentioned is not great so we have our our Spark uh application here it needs to read up the data from Park files from The Lakehouse needs to deserialize it we do a column uh row pivot we do our join with this inefficient um uh point and time join in spark or Hive um then we need to Pivot our data back again because we're going to send it as jdbc reoriented data to the to the client and we're going to serialize it and then the client we're going to deserialize it and then pandas wants it in color format so we need to do row column pivot again we're using SQL Alchemy for this there's a couple libraries a little bit better but it's still um you know it's it's not high performance let's just say now there's another thing that this service does for us is it can also create training data so sometimes when you're doing when you're when you're training models uh your Panda client isn't able to handle the data volume you maybe have you know billions of rows and um that's not feasible to return as a pandas uh pandas data frame so you can uh materialize that training data as paret files or CSV or TF record um so that's another thing that that that we're doing in spark and we can also do now in duct DB so this is our new architecture it is for reasonable scale data as I mentioned so whatever the memory capacity of the duct DB server is that's typically what we would recommend is the amount of data it can return duct DB does um it does spill to dis in the latest version since I think 0.8 um but that's not really super production ready so we recommend keeping keeping the amount of data returned to to less than the amount of memory in that server and if you want more you can still use spark and but you can see here is that when we read the data H into our server we're reading it directly from the paret files as Arrow data and then we return then we do the do the operations that we need to do maybe we can do push down filters as well to the uh paret files maybe we're going to do our join in here uh point in time join and then we return the arrow data directly with arlight protocol to the client and the client if it's a pandas two client can do a zero copy of the arrow data directly into the data frame and you're good to go um this is pretty cool so let's look at the challenges and I'll show you some of the numbers at the end okay so um this is an Enterprise service so we're an Enterprise company so we have to kind of care about all the Enterprise stuff so I'll go a little bit through that so firstly we needed to do access control and authentication you don't get that of course out of the box with the open source platform um we're using Mutual TLS for clients to authenticate themselves with the server and we have access control lists uh inside in hopworks for the different projects that people are members of that was all great um you know at the back end we do access control on our we have a file system there on top of the S3 where the where the table format is ultimately stored so um to to get access to do that access control within our our file system we needed to uh duct DB to be able to read and write to our file system now our file system is slightly different right so we are storing the data ultimately in a book at an S3 but we write to that file system via hdfs API we have our own file system called hopss and it plugs into S3 and the benefit of hops FS is that it's a like you can think of it as being a uh globally consistent cache across all of the workers in your cluster and you can read the data from there from that cache uh with nvme speed so instead of having to go to S3 and read it 150 megabytes per second whatever if you've got an mvme disc there you can read it potentially gigabytes per second um so we had to plug that into duct DB that was pretty straightforward because there was a connector for it that's fine um what we also had to do was uh write a connector for hoodie so one of the challenges with the these Delta with the lake house file formats currently is python compatibility Delta lake has a python client called um Delta ORS if I remember correctly um there's also one uh in the works for for Iceberg called Pi Iceberg they have read support now but WR support's not quite there yet it's on the way and and uh hoodie is a little bit after right so there's neither uh any support for reading or writing with python currently so we had to implement reading um we're using hoodie um copy on write tables uh so it's a little bit easier than the merge on read tables in hoodie and but I'll show you a little bit details how we did that so um hoodie tables uh consists of many different uh parket files and there's a lot of metadata related to them so every time you do an update to your hoodie table you get a new commit and new metadata uh potentially stored in different partitions and when you want to read your data um in our case we want to read the latest value of our features so the latest commit um what we would do is we B basically need to identify these Park files that make up that latest commit so um we did this uh efficiently ourselves and this is not something we've released publicly because we're not super proud of it but basically we're we're um we're identifying the files with uh using our file system to do lists on the files now that's um that's something that's very efficient in hops FS it's not something you would do on an object store because listing in Object Store is super slow and you only to get back a thousand files at a time U but in hopss it scales um order of magnitude faster so that's how we get get the uh current files that make a the commit and uh then we need to to basically serve multiple clients in parallel so if we have we have a servess platform with 4 and a half thousand users what happens if many of them are are reading data at the same time from the same same arlight server um we need to somehow you know manage this so one of the things we do is we set memory limits in duct DB that if for example a user tries to read more data than than is available on that machine we'll you know throw an exception and tell them hey that was too much data go and try and use spark instead that's pretty straightforward um the other thing we need to do is that we need to handle multiple concurrent clients reading so in what we do here is because each arlight server is independent we can trivially load balance the requests across them so just do a random load balancing uh pattern across those nodes and we do that um uh in the cloud so that's pretty much it and these are some of the numbers that we got out of it now these are public uh benchmarks they're uh for both the online Fe if you're interested in feature stores as a platform a way of getting data for training and inference in your models we have uh the feat store consists of two layers you have this columnar store the the file format or sorry The Lakehouse format that's what we call the offline store but there's also an online store with low latency access you can get both of them at that uh GitHub URL and you feel free to to to like it if you are if you're there and but the this this is one of the benchmarks this is The Benchmark for the offline store um so it's a very well-known data set called the New York Taxi data set which is about uh taxis picking up passengers and dropping them off and you can see here we have three tables and we can do a point in time join across those tables to create uh consistent training data app point and time join is where you have you start with a prediction Target or label which we have in the Raw data here so we we we have a model we're training about estimating the fair amount for example and we'd like to include features from the pickup features and drop off features in the training data set so we're going to join them but you can see here we have the observation of the uh fair amount and we want to know those values for these particular features at that point in time so what we do is we go back in time and we find the first the most recent uh value for that feature at that particular point in time and we can see this for Feature A and B here and that's what the joint is doing so this is some of the numbers we have comparing uh hopworks with sag maker vertex and D feature store and um it's pretty big right there's experiments in the get a repo for for the different size of rows from five to 50 million they're pretty consistent across all of them um you know as a as a python developer if you're using a feature store from one of these Cloud vendors and you want 50 million rows you know you can go for lunch and then come back and you get it as your Panda data frame it's not exactly super uh great user experience um but in hopworks now you can you couldn't get that back in you know 30 seconds or 20 seconds which is pretty it's a different experience the Chang is how you can work with data and particularly for data Centric AI you want to iterate faster on changing different combinations of features and and and filtering out data and so on this makes it very much an a platform that you can work with uh interactively the other thing you you might want to do is your data might be bigger and you want to save your training data as files paret files this an experiment again we can still see a huge performance input not quite as big in hopworks but still much more significant so here for example dat kind of works in this case except for it didn't work a 50 million um but you know so spark Works a bit better so it's it's relatively okay at at generating uh train data's files compared to uh you know in spark the the problem they had of course with Ds here is they're still doing this two pandas on the driver which tends to blow up so that's pretty much it um you know what I talked about here was how we want to improve feature store performance so we can get this python Centric um into interactive development uh model working with feature stores we want to basically help improve performance so performance actually is important in changing how you work with a feed store platform um it should mimic your local workflow and right now we're getting you know pretty much line Network line speed reading of data from the feature store but our right path is still slow so that's the next thing we're going to work on and the other point about this is that you still do need something that's really scalable for the really large workloads and and we still have spark available for that that's it so if you're interested in find out more um you can go to visit hopworks um there's also featur store.org that I I I organize which is both a website and a conference um and you can find out more about them there thank you yeah we have a couple minutes for any questions you can asking up to there's one go there so with the increases in speed and decreases in latency with this rewrite do you still need a distinction between like an offline feature store and an online feature store and why yeah it's a we get this question a lot it's it's so the the basic idea of a feature store so there was a feat store called spice machine which is based on um you know this hybrid um uh uh oltp olap so it basically does both so you have these hybrid column or row oriented stores there's a single stor as one of them for example um the reason why they haven't been really successful is that there are a lot of use cases uh I'll give an example one personalized recommendations that are just really low latency so if I'm doing personalized recommendations and um I would go to you know I would say hey I've just you know clicked on this link or I've typed in some search query um what will happen in a in a kind of a state-of-the-art architecture is it'll go to a a vector database it'll get back a few hundred candidates then it goes to the feature store with 300 or 200 rows uh primary keys and say give me back 300 rows or 200 rows and I need that in 30 milliseconds or 40 Mill seconds and the hybrid um stores Coler ra oriented stores they just I mean their order of magnitude off there so um typically people you know and this is a general Trend in we see this that data volumes have being increasing for 30 years massively and more specialized stores are just appearing all the time and people said you know graph database I got spark and graph x why do I need it but you know it's a billion dollar market now so I think just as data volumes increase there's still always space for specialized databases that handle um you know popular workloads that that that there's demand for