Fulfilling Apache Arrow's Promises: Pandas on JVM memory without a copy

Speaker 1 [00:02]

So my talk about today is about how we have really data in the JVM And we want to work with pandas on the top of that so it's a bit of crazy more far-fetched I'm normally doing error talks, but as there was already a bit of error in the keynote I thought I have to write something more crazy and Part of the first time of the talk was already working on error data In pandas without doing a copy that I did that talk in Berlin this now is one step even further even more crazy I have data in arrow in Java, and I want to work with panels on top of that, but not doing any copy Sadly tomorrow this morning my internal keyboard Brooks. I'm also doing now with excellent keyboard a presentation So I'm not sure if the code presentation at the end will work as smooth as I hoped But beginning I'm its data scientist at blue yonder I work at my play on a time on Apache arrow and Apache pake we use that already productively and but in the end, I'm a Pandas user, so everything I want to have at the end should be accessible in Pandas, should work fast with Pandas, so that's always my aim. A nice thing, we have some disadvantages in Pandas, like it's fixed Python, and that's a nice thing which happened in 2016. Apache Arrow was announced. The basic core of Apache Arrow is that it's an in-memory data layout. So it actually specifies how you should place data in memory. Not on disk or so, you can put it on disk, but the main thing is keep it in memory. And by keeping a sane and specific specification in memory, you can pass it around between processes or over your host, and you don't have to reorder anything. All the participants who understand Arrow can work with the memory directly. It's a common memory layout, so it needs to be catered for a lot of cases. So even though defining a memory layout sounds simple, you need to keep in mind how to deal with the SIMD instructions in your modern CPUs, how does the cache of a CPU work, and also a bit in mind, how does it work on a GPU. And one of the really good things to see how well the BirdShark is running is there is support for Laven languages. The pull request for C-sharp is open, or maybe even merged by now, so that we have really a long list of languages. Hopefully I didn't forget one here. But it makes it really real as a universal framework which you can use everywhere. As a Python user, there are normally two typical use cases where you know Arrow. One thing which got me started is I wanted to read Parquet files in Python. And I wanted to read them really fast because I have huge amounts of data. So waiting for them is not really so nice. And also, the big public experience of Arrow, or where a lot of people will use it, is there is the pandas udf decorator in spark you can use that to make spark udf functions written in python much faster it's called pandas udf but in the end what it does it has a serializer internally that uses apache arrow to talk between scala and python but you talk with it over network so you're copying some data and in the end copying data is not what i really like because copying data takes time, costs RAM usage, and that's all the things Apache Error in 2016 said, yeah, we're going to solve this. These are two main pictures, which were on the homepage in 2016. They're still on the homepage. They're still the goal of Error, meaning you have a lot of data analytics products on top, and on the bottom, you have a lot of different storages. You want to bring them together, and you don't want to write every pair of communication between them, custom serializers that are slow, that are complicated. But when they all commit to one memory structure, at least for interchange, you write once an error adapter and all other parties which also implement error can deal with that data. And they can deal with that data really fast because the serializer, you write one serializer or one conversion or you already use error internally and you can concentrate on that. In 2016, this was only a goal. but we're actually at the point now where depending on which two pairs you pick it really works looking about how you do machine learning or how do you do data science at the moment a typical workflow for me at work is you have an SQL engine maybe backed by a normal relational data store maybe it's backed by some parquet files or maybe even something else but you have an SQL engine that you use that to join your data to get a big table view flat view that you can put it into pandas use them pandas for data pre-processing shuffle around some columns fill some missing values and then push it into your machine learning model the last two tasks they happen both in python and they're really efficient but there's always a kind of disconnect because this workflow here It looks simple, but in the end, this isn't simple workflow. In the picture, just three components, simply working together, but between the SQL engine and Pandas, there's normally a big delay. The engine is fast, Pandas is fast, but the layer between isn't fast. That's mainly due to that everything in these data landscapes, which data analytics, data science, is really heterogeneous. There are these big Hadoop setups with Java products that work really well, at scale okay but they're in the java world and the machine learning we typically use python which is a different whole whole ecosystem other data structures and so on so when we look again at the workflow we had just um there's a whole gap in between we have the sql engine which then returns data through driver if it's in java in my case here now what i'm looking at it's a jdbc driver so there's often databases bringing odbc drivers but normally just yeah here you have a odbc driver it's just enough that you can use tableau on it because tableau or something else needs odbc but we actually don't care use jdbc because you typically also have a java service and also often you have to speak engines at the end you normally get a single line result because So you do an aggregation of one million or one billion rows and in the end, you're interested in your report to put the whole profit of the year, that's it. But for making data science, we normally want all data. We wanted a big, flat view and it's a huge amount of data. That's normally what drivers are not optimized for. And in Python, you can use JDBC drivers. There's a thing called JDB API, which, Which starts at JVM in the background, loads the JDBC driver, and then takes the JDBC rows, which are returned by the JDBC driver, and converts them to Python rows. Sadly, Python rows is not the data structure we're using in Python, or in Pandas. In Pandas, we want columns of data, especially columns of NumPy data. And we all know that Python rows are really, really slow. And when we just make this conversion, we have to wait a long time. In Congress, if you do that conversion in Java, we have at least a just-in-time compiler. That's not a thing we have in Python or Pandas. And also, we're copying the data here several times. We have it as a row form in Java. We copy it over into Python. This is the most expensive part because we copy it and we have to cross the language border, which costs us a lot of CPU power. And then once we copy it again, when we pass it to Pandas. now the nice thing in the arrow project is i talked about that we have 11 languages support it started all out with having the c++ and python implementation and on the other side the java implementation i think at current state they're still the both with the best support and we're normally looking at python conference at the python and c++ side but actually as we have data in Java returned by JDBC driver we can use a tiny bit of Java and the good thing is in arrow we have since two or three months a JDBC adapter which gives you anything which is JDBC query returns then as an arrow record batch or in Java it's called a vector schema route which is basically the same thing as a panel data frame collection of named columns record batch and in arrow is the most simpler structure, it just contains continuous arrays. There's no chunking or slicing happening. We can do the row-to-column conversion in the JVM. That's sadly a thing we always have to do because JDBC only has a row-wise interface. So maybe a better thing in future would be to have an alternative to JDBC that already gives you columns. The other nice thing is Arrow uses off-heap memory. which is not always the best thing in Java, because Java developers are a bit scary about off-heap memory, but you can use it, and Arrow gives you nice interfaces to it. The other good thing about it, it's the same memory layout as you would use in Python, meaning that the data is already in memory how you would use it. You could pass it to any other process, which gives us then an updated workflow, where we have a JDBC driver, we have this error adapter, and we have an error table. and the bad thing is we have an arrow table or a record batch in Python in Java but we're still not in Python we can't access that memory and also we have arrow and at the end you want pandas and these two are still components that are not working fluent with and fluent with together already mentioned at the beginning there's PySpark PySpark has a conversion layer that you can write some data in Spark and then pass it over the language border to Python. This is already giving you a good speedup, like sometimes two times, sometimes 10x. The disadvantage is still it copies data over network, meaning you have at least one data copy. And network is normally not so fast as if you just take the memory as is. And also the error-dependence conversion is really fast, so there's a lot of magic happening in there, and that's a good thing that we have Wes in the project because he also knows the Pandas internals. So we directly fill the Pandas data structure as Pandas uses it internally. We're not giving in a dictionary of numpy arrays, but we're actually giving the data structure Pandas is using internally, coming up to that later. Now to look at the data where we have. We have an error record batch in Java, and we want to use it in Python. The nice thing is the layout is the same. So actually, I have to just get access to that data where it's in there. And we have a new module in PyArrow which just does exactly that. It takes the Java objects through the same code that we're using in the JDB API, which is Jype, which gives you access to a JVM running in the same process. but instead of copying the data over it just accesses the Python the Java error object and makes the new metadata and Python out of it. And it's just the metadata it's not taking the actual payload data and the metadata is just typically in the range of some bytes or one or two kilobytes and you can have 10 gigabytes of data and no copy of the actual data will be happening and we pass these references around We also have to be careful, but the module is done, that we take care of reference counting because we're now having Python objects that reference Java objects that reference our memory from error allocated in Java. And that brings us now to the point where we can actually access the data in Python and have them done a copy behind that. So the next step is we should bring it into Pandas. But sadly, pinus is not the same thing which we actually want to have. An arrow table is a collection of arrays with a column name. And inside an arrow array or a column, there's one array. But this one array is at least two memory regions. It's a valid bitmap saying if there is another value in your actual row or if it's not. And then the actual data. Pinus in itself, it's made up out of NumPy. NumPy has not really, it has some support for missing values coming through mask arrays, but mask arrays are more second-class citizen in NumPy. And Panas itself is not using these mask arrays. It's just using standard NumPy arrays. And that's a typical case when you have an integer column, you assign none to it, and it turns into a float column. So the same thing is when we have an integer column in Arrow, you assign, it has a none value in it. and when we want to convert it to pandas, it will turn into a float column. So we have to copy this column. The other thing is there's a pandas internal thing called the block manager. The block manager allocates all data that is from the same type into a big pandas matrix or NumPy matrix, meaning that if you have 20 float columns, it will be a 20 size number of rows matrix in your pandas data frame, and there you store your data. And in Arrow, you normally just take columns. So when we have more than a single column per type, we have to copy. Yeah, so we have the data types which don't match. Also, because there is no string type or there is no decimal type in pandas. In these cases where you have those, you have to copy. I already mentioned the block manager. Pandas groups by types, so that's also a thing where we can't just copy into pandas from arrow because in arrow, columns are separate And nullability isn't a problem. So at the end, there are at least three points, maybe even more, that means arrow memory and Pandas memory are not the same. So you have to have a copy, and copies are bad. The cool thing is if you have like five gigabytes of data, if you have it on your laptop, your laptop has 16 gigabytes of RAM, you still can work with it. But if you do two or three copies at the same time, your laptop is too small. a nice good feature in pandas which is available in the latest release but actually you should wait to the next major release of pandas are extension arrays extension arrays are a small set of interfaces you need to implement but they're really powerful the powerful thing about them is you implement storage for pandas column but the analytic on top is provided by pandas and not by you. You can overwrite some of this analytics, but the magic happens all inside of pandas built on top of you. And in 0.24, there will be much, much more features building up on that, like the categorical columns in pandas or the timestamp columns. There will also be extension arrays. And the really nice thing, people are looking forward, there's also an integer extension array built into pandas in 0.24, which gives you nullable ints built into pandas. but this kind of concept is something we actually could also use because we have already Arrow data which is stored in memory which has the same properties so I created a library which is a bit more prototype not production ready it's also a library which needs kind of a lot of master branches or per requests it's called Fletcher it implements the Apache Arrow it uses Apache Arrow as backend storage but it implements the extension array interface and to keep it pure python i used number on top of that to have the missing algorithms implemented so there are still two or three things you need to implement which why not i get an arrow and just to give a small prototyping and fletcher i use number on top of that you can pip install that but i would suggest always to use the master branch if you want to try it and also the best things also if you use arrow master pandas master with that but that's also maybe a future thing which could be built into pandas just using arrow as a back-end storage so that you don't have to always copy your data and the benefit of um using that is now when i pass all my arrays i have already in my record batch coming from java i can make in Pandas DataFrame out of the record batch simply by allocating these extension arrays. And at the end, I have the thing, I have data which comes with my SQL engine. The JDBC driver talks to my engine, gets the data, sadly still converts to JDBC rows, but then I have an arrow converter which converts the rows as my own single copy to arrow. The arrow memory is allocated in Java, referenced in Java, but it's passed in over language boundary just by the metadata and the memory addresses, and then referenced also by Python. And in the first case, just by PyArrow, but by using also Fletcher, I can turn my arrow columns into Pandas columns, and I can actually work directly with Pandas on this data, and pass it down into my machine learning model, meaning I have just a single copy between JDBC rows and arrow data. And this one copy is sadly unavoidable, but it's at least not on language border, it's still on my Java side. So if you have building Java databases, which probably not so many people in here will do, just think of just returning columns instead of rows. It's really good for analytics. And also, a lot of these things I'm telling you here is a lot of big pictures, so the question is, is he actually telling you something that's working or is it just giving you big pictures? I'm not going to show you how the code works for the non-arrow side. It's quite simple. At the top, it's importing JDB API. It's connecting to the database, giving you the driver. I've used in this case for benchmarking Apache Twiddle, which runs really good in a local mode, and I just use one big package file just to have the machine learning expertise. I have this one table which is really big and I just want to fetch it and JDB API is implementing DBAP2 so I can just pass and read SQL and pandas the main problem why I'm not showing it it takes 11 minutes on my laptop is an awful lot of time just for loading data and that's already on my laptop if you would pass it over network will take even longer and and we can now also have a look at how this will work when we're not using JDB API but when we're using the code I mentioned here. I have to do a lot of of imports, and I'm using here a small snippet called IPython memory usage, which shows me the memory usage of my current process and how my current cell affected that. So that you actually see what's happening inside here. So I had already my JVM running, my Python running, and also my JVM running. That's why you see a large decrease in memory, but here, the next row we see, we're clear with memory, and we have a current usage, which is quite high because we have a warm JVM, and a warm JVM in the background takes a lot of memory. But the thing is now we want to get to the point that we see that we don't take any more memory. So we can load a driver, allocate a root allocator in arrow. That's basically an allocator for arrow tables saying you don't allocate any more than this amount of memory. As we're not building a database system here, I don't care about that, and increase to the maximum size. connect to my local database and use the same query I showed on the slides again which just selects a huge amount of data and execute that directly with Jive. So this code here is connected to the JVM which is running in the same process but another thread and calls the SQL to Arrow adapter which takes a query and takes a root allocator so it can allocate arrow data and pulls all data from the database into here an arrow JVM memory. At the end we can see it takes 293 megabytes, so there's already some data in here. The next thing is I have the data memory, but I now want to have it in Pandas, or actually I first want to have it in Python. I can then call this JVM record batch which takes a JVM record batch and turns it into a Python record batch in Arrow small bit of metadata added 0.4 megabytes it's okay it's a bit more than I would have expected but it's a lot less than my 300 megabytes above and now I'm using Fletcher which is shortened with FR here to convert my Arrow record batch to a Pandas data frame and that also gives me a bit of increase because i have now a pandas data frame object but it's 0.2 megabytes not 300 as above and to show you that it actually works we can have a look at the data in pandas and can see here we have some data in there pandas can plot it and we also can call bf.info and see that we have typical pandas data frame output here at the top but also we have this kind of magic fletcher columns which are the same types we used from NumPy also but they are fletcher columns because in the end this is not a NumPy type at the end, this is an arrow type and you can work with this data here and yeah, that basically shows how you can use JVM data in Pandas I hope that was convincing now that it actually works and it's not just some fake I put on my slides and if you're interested in these things I work at Beyonder, I have fun with that join me or the other important thing is Apache Error is an open source project there's a lot of things going on and there's a lot of things where you actually get this kind of 10x or in my case it was at least more than 10x performance improvement it takes some time to look into the project by actually to get it's right satisfying at the end because you get a huge amount of speed ups and you also get a huge amount of possibilities because you can now connect systems which were totally separate before. So thank you for that great talk. Any questions? so thanks for a great talk very interesting I was wondering about that Fletcher project are you one of the maintainers or is it just something you discovered and if you are actively maintaining what is kind of the vision for it so is this something that will go away once Pandas and Arrow are better integrated or is it something that will remain in your point of view I'm one of the maintainers so the other maintainer is Florian who's also at a conference here and will give me a talk tomorrow and also Christopher is one of the persons who is also at a conference but also gives a talk tomorrow so all three maintainers are here at the talk or not at a talk but at a conference the thing is the prototype at the moment it's not something you can use in production it's more kind of testing a vision because there's a lot of documents from Wes McKinney where you can look up what's his vision for Pandas 2.0 and one of those things is build Pandas on top of Arrow which is a lot of breaking things because you're not using any more NumPy. But Fletcher is a good thing to test. Can you work on top of Arrow? There are a lot of things still missing and it's not really working perfectly. Or in some cases, it's really slow. But there's a lot of cases like when you can already take Arrow string columns and that's a thing where you already get a two to four times improvement. But yeah, maybe one day it will be also part of Pandas. Any more questions? Otherwise, I have one. We just heard a talk about PyTorch. Do you know if there's any efforts to also connect deep learning frameworks to Arrow? Uber has released two weeks ago a library called Petastorm, which connects Arrow and Parquet to PyTorch and TensorFlow. They're embedding videos and images into Parquet files or Arrow files, and then use PyArrow to load the data into Torch or TensorFlow. Cool. Thank you very much again.