Scale By The Bay 2019: Jacques Nadeau, Vectorized Query Processing for CPUs using Apache Arrow
[Music] thanks so much appreciate it so as we mentioned Jacques it's nice to meet you all of you I've got really two hats and I'm talking I'm so to move between the two different hats in this talk so one of my hats is that I'm co-founder and CTO at Dromio and the other is that I am the PMC chair of Apache era and so I'll be talking a lot about apache aero what it means how I can make things go faster but also give you some context in terms of how we use it at drum you know to build our technologies so quickly I'll go through the background on arrow and sort of how it came about I'll then talk about the components of aero I'll talk about sort of the basic data structure which is the aero data structure so you got a little bit of so that in your head and then I'm gonna go in and talk about Aero specifically for query processing and how we found different techniques to make it perform very well for processing purposes and so that'll be hopefully a quick 25 minutes or so so to start with most of you have probably heard of aero by now I hope it is standard for commoner in memory analytics processing and transport okay and so it is focused on many different workloads trying to speed up those workloads designed to allow you have a common day later between different components it's designed to work with any programming language that you want it to work with and there are quite a few that it supports already and one of its core tenants is impatient project and so in Apache foundation projects at his core protects you from being locked into any particular vendor or anything like that and drives and this driven entirely as a consensus driven project so where did arrow come from well it came from a collaboration between myself and Wes McKinney we were talking about different problems we were dealing with in data if you don't know Wes McKinney he created pandas he wrote the most popular O'Reilly book ever a brilliant guy and we were we were chatting about sort of data problems in both data science and in data processing databases and really we kind of came to a shared conclusion and the conclusion was is is that one of the big challenges when you're building a data system and what I mean by data system is like what a company needs not what one particular piece of software provides right because generally speaking when people are working with data they need lots of different technologies and you'd be a bring those together to solve actual data problems one of the things that happens is it becomes very expensive to move data from one system to another okay and so typically you'll try to do as much as in one system as possible before you go to another system it's because the boundaries between systems are expensive to move through and so we started looking at that problem and said you know what we need to figure out a way to solve this and so this kind of started out as a challenge around transport it's like how do I move data efficiently between systems and so there's actually lots of transport approaches to data already like I could use rest and JSON I could use protobuf or thrift or something like that to move data between systems but what we started to identify was is that the core problem with all of those existing ways of moving data between systems was is that it was expensive to move from what would the internal representation of data was in each data system to what they could put on the wire and then communicate to another system and there was different reasons for the expense one was a transformation of the data in memory into a format that could be more appropriately sent on the wire another thing that was a problem was is that most of the ways that you expose data was through api's and so if you wanted to interact with a particular amount of data you actually have to call some set of api's and whatever language you're working in to access it cell by cell and so it basically means if I want to read you know ten records and I mean columns in each record I've got to do a hundred method indications just to read that data right and that was kind of the pattern with everything that was out there and so we came to the conclusion with the way to solve transport and so this this sort of serialization deserialization boundary was to actually solve processing first the idea being that if you start to use a common format inside of your processing engines then when you decide to move from one processing engine to another you don't actually have to change the format you can take that format you can drop it on the wire hand it to the other system and actually let that system process it as well and so really this concept was hey in order to solve transport you actually need to solve processing and so arrow was designed as a processing first representation of data that also is well designed to be able to transport between different systems and so it's kind of this chicken and egg thing about how we got to the idea of arrow in the first place and so arrow has done really really well so arrow started about four years ago three and a half years ago something like that I don't know exactly and we've seen huge amounts of downloads at this point lots of different echnology is adopting it and so I think that it's here to stay I think we hit past four million downloads a month about a month or two ago so very happy with all the different technologies that have adopted so I'm gonna do one slide about dromio because that's people who pay my my paycheck and so quickly we built a technology on top of arrow called dromio which is a data Lake engine we describe it as a processing and semantic layer to interact with data wherever it might be stored and it's all built on top of arrow everything that I'm talking about in terms of the technologies here we use that we use those things at hundreds of customers running on thousands of nodes so they're all things that you can trust his production already stuff so I'm gonna I added one slide to my talk because I was listening to Thomas's talk just before this talk so those of you that were here with Thomas's talk from from Salesforce I actually very passionate about what he talked about which is a table format and so one of the big missing pieces I think in data Lake suspiciously has been a table format a common table format that can be used across many different systems there are actually several different technologies that are in play there and we actually have also come to the exact same conclusion that Thomas did which is does the iceberg is the right one to follow and so I strongly recommend to each of you if you haven't thought about deciding on what table format you have and sort of sort of focusing on investing in that you probably picked one of the others and I'm gonna caution everybody that there's a path for all of the others that's taking you down a path that is going to feel more and more like lock-in because most of the other technologies are beholden to a small number of companies or a small set of technologies and not really designed to be consensus driven to work with a large set of technologies that's that's my opinion but it's something that I strongly suggest that you look at if you're thinking about how to do tables and if you're not thinking about you should probably think about it anyway off-topic but it was something that came up when Thomas was talking so I want to bring that up so let's drop back into arrow so what are the components of arrow well the number one component is is that it's an in-memory representation but like memory by itself is not something that people can work with people want to program against api's and so in addition to that there are actually I think over a dozen now I have some of them here but I believe that there's over a dozen different language bindings for arrow representations so each different language binding figures out the optimal way to interact with the arrow in memory representation of data that is best for that language right and so one of the keys of the arrow representation is is that you never have to call into another kind of another language in order to interact with data representations you can always use whatever's appropriately native to that language so on top of that there are lots of different what I prescribe is building blocks so first you've got the memory format and then on top of that you've got sort of bindings for different languages and then you've got these what I call building blocks okay and so this is for examples there are multiple other examples of building blocks that are inside the arab project or near the arab project okay one is Gandiva an I'll talk more about this but it's an LLVM based just-in-time compiler for expression evaluation another one is feather which is a non disk format for a femoral data movement between two different applications arrow flight I'll talk about that as well an RPC protocol for moving the arrow data between different systems and park' which you probably all know of we've actually done a huge amount of investment to make park' and arrow work really well together and just to note quickly there one of the important things to understand is arrow is an in-memory representation park' is that on disk representations so they don't compete at all they're actually very very complementary and they both have similar concepts and that they benefit from using a common representation but the common representations while complementary are not the same and then on top of different building blocks there are lots of different integrations that people have built on top of arrow to use for their own purposes right so pandas is actually using it to move data between a bunch of different types of processes SPARC is using it to move between Python and panda since Park I don't think it's enabled by default but if you turn it on it makes things like 50 50 or 70 times faster if you're using those kinds of operations join me as I mentioned we built our what we call our Sabo engine and on top of arrow and then nvidia actually they have their Rapids initiative and the CUDA data frame and that initiative is actually all built on top of the air representation as well so it's not just about CPUs it's also about GPUs so if you want a high performance representation of data for processing purposes this is the one that you want okay now before we go any further one of the things I think it's really important to cover is what is arrow and what isn't arrow okay so people will kind of hear the words and they hear a lot of different ways that arrow could be used but I want to make sure that people understand what it is at its core it's a specification for how you're gonna have data in memory it's a specification for how that data is shared between different processes and it's a set of libraries and tools and building blocks to work with that okay it isn't an installable system you can't go and say yum install or brew install well you can say brew install arrow for the development stuff but you can't you don't install something you don't run a service there's not an in-memory grid that is built that it provides it's actually an internal tool for solving data processing problems okay it's also not designed for streaming applications is designed for analytical applications we're gonna be moving large amounts of data okay if you're removing one record at a time it's not going to work well for you so quickly the structures around arrow when we describe it as a shredded nested data structure it's designed specifically to be randomly accessible so you can get to any record without having to go over other data structures and this is one of the key differentiations between arrow and memory data and something like park' on disk which is not designed to be randomly accessible okay when you're moving data between different systems you're generally following this message pattern where you basically negotiate schema between two different sets sets of processes or sets of parts of your application you then communicate one or more dictionary batches so use arrow does support dictionary encoding and then you send what I what we described is record batches which is a batches of data that are in column two representation that are typically not huge so that you can pipeline between different parts of the system they can be arbitrarily large but generally for pipeline purposes we don't make them that large and how does the data actually lay out in memory well it's pretty straightforward it's pretty obvious but it's designed in a couple of ways to make it really efficient for processing and so let's imagine that we have some data on the left hand side we have some an array of persons and each person has a name and age and an array of phone numbers okay that data represented an error format is going to be what's shown here on the right-hand side which is that you're gonna first have a name vector okay and a name vector or name array is has actually two sub structures the first is called the offset array which tells you what the positions into the second data structure the values array actually compose it points to so the first record is going to look at offsets 0 and 1 and say the start is zero offset zero and the on the end is offset one and that means that we're actually getting 0 2 3 is the first value for that string so that's Jo E and then the second value will be start at off the second offset in the in the offset array which is gonna be value 3 and then the third offset which is gonna be value 7 okay so 3 2 7 is actually the next value right those are two completely independent memory data structures and so for example if you wanted to process understanding the lengths of strings or be able to do a first-level pass on a quality of strings by using lengths you can interact only with the offset vector and never actually interact with the data vector okay age let's say that that's a four byte integer that's a very simple representation it's basically an array of four byte integers and end ok phone numbers is a list of strings okay so it looks kind of like the string did in the first column but you actually add an additional list offset to describe which elements in the individual string which is in the individual strings are in each list okay and so the design of these things are all these sort of array based data structures and that allows you to do a bunch of interesting performance things so I'll get into so it also allows you to have a very clean way to pass data between different systems now I didn't actually put it here for simplicity purposes but every single structure inside of arrow is also possible it can be a null value or not an unknown value okay and so in addition to these data structures each of these things has a bitmap which describes whether or not each value is valid or invalid okay and so when that data all gets laid out in memory we've got that same data structure here we actually compose to this into a record batch and so here we have columnar data I've lost my cursor so I won't use my cursor but we have columnar data that describes basically the three structures for named I just talked about the two that I talked about plus the bitmap we have the two structures for age and then we have the four structures for phone numbers okay and those are just all laid end and in memory you can do a gathering right on to the socket communicate those things and then you put a very very small data header above it which describes what the offsets that the lengths of each of these structures are okay and so that's the amount of data that you need to be able to communicate between two different pieces of software how this data works okay and so if you think about that example I was talking about before where you say I want to send her ten records with ten columns from one process to another I would have to run a hundred method implications well in this case I can hand this chunk of memory from one process to another process and I have all hundred hundred values without having to do more than like one method invocation right and so not only do I not do any work in terms of serialize and deserialize in between moving between systems I also don't have to do a lot bunch of method invocation x' to get access to the data in this in a system that's receiving the data so now that we have that data structure let's talk about some ways that we can process that efficiently so the first time I talked about is a project called Gandiva okay so Gandiva was something that we built at Dromio an open sourced and contributed to the Apache era project and it's designed really to be able to allow you to process aero this memory representation of data very very efficiently okay so if you get data into that in memory representation now you can start to do really interesting things with it and so we have seen substantial performance improvements in our use of it and have it in a bunch of production use cases so let's talk about what it is so if you think about a data processing pipeline you're basically taking data and moving it between you know between two different points right and so in arrow when you're doing processing of arrow you're generally dealing with these arrow record batches and you're moving them through a pipeline okay and so the design of the Gandiva execution kernel is to be able to take an arrow batch and turn it into a new arrow batch and that's specifically done to apply an expression tree against that okay so a very simple expression tree might be a plus B and so Gandiva understands the internal representation of arrow and then optimizes the instructions that are designed so that you can take advantage of that representation so if you normally wanted to interact with data you would use something like a cell level interface and you would call into that and you couldn't take advantage of how that data is out laid out in memory because the data in memory and arrow is laid out in memory and aligned that means that you can actually write the code for processing to take advantage of those those properties and so as I mentioned it's an expression library it has I think hundreds or maybe even thousands of expressions now that you can apply against for different operations here's a couple of examples that you can do at the bottom here and so let's talk about how it can make things faster right and so I talked before about you've got that validity bitmap which tells you whether or not something is true or false and then you've got the data itself that actually describes what the values are and so if we use the example of a plus B okay when you're trying to actually do a plus B in a processing system let's say a sequel system like Jerome ium you actually have to do several operations so first you actually have to apply null semantics so if a is bead and the outputs actually B I'm sorry the outputs no if B is null then the output is no okay so if either of those values is no then the output is no if they're both valid then we do the operation okay and so if you normally would do this processing you would have a bunch of if statements to evaluate those that the validity of these values and then you would then do the math if you needed to okay the problem with that is it doesn't pipe well PI playing well on a CPU because you're not going to necessarily be able to guess whether this is going to be true each values going to be valid or not valid okay and so once you know the representation and you know that the validity bitmaps are bitmaps and you know that the data is stored independently you all a said and can actually decompose this operation and do it in far less instructions okay and so in this case we can actually take the validity bitmaps and do a bitwise and of those maps to determine what the validity of the output values are and we can do that at the word size rather than at the individual record size right and so I can do 64 values at a time do that and and reduce produce a new validity I actually didn't do any if conditions at all to decide whether or not to do this operation and then I on the data side I just always do the math okay so again well it potentially means I'm doing the math more often I completely reduced all the other instructions that are happening so it's actually substantially more efficient okay and so we have these concepts of how we decompose all of these operations because validity checking is actually a huge amount of expense in all this processing that so that's one example of how it goes faster another is is that we've lined up this data to be designed to work well in modern CPUs and so one of the things that modern CPUs have is they have 70 instructions or single instruction multiple data and that means that you can actually interact with multiple independent values simultaneously with a single instruction and so here let's say that I've got you know for two byte values two byte values that I'm adding to for other two byte values okay normally I would do that each value at a time but with simply instructions I can say actually add all these at once and then get a new result okay and so on the if condition side I've reduced a bunch of instructions then over here for addition I've also reduced a bunch of instructions and so these are different ways that are techniques to get more processing getting more done on your CPU in less time once you have that representation and you know that representation well there's a lot more to this a lot more detail you can go and look at it in the code but that's kind of a high level on Gandiva so how does it work well basically we defined a again language agnostic way to express what the expression tree is so you can describe an expression tree using protobuf okay so any language can describe this expression tree and then you hand it to the arrow library from whatever language you're in it compiles it using LLVM at runtime to build the actual transformation logic and then you have different language bindings to be able to push data through the the Gandiva core okay so you can basically like whether it's and so we use it in Java we also use it in C++ other people use it in Python there's a breast bindings I think now as well so there are lots of different bindings to be able to push stuff through and process it efficiently so how is it actually impacted us well in a lot of cases let's say a really simple expression tree it's not that much difference right a lot of systems are actually pretty well optimized for simple expression trees so you know 2 to 5 X something like that but if you get to more complex expression trees which frankly in the real world when you're working with customers that have giant you know 5000 line sequel statements you frequently see a huge amount of additional lift and so I've got a branch mark link there which is one of our customers scenarios where we applied Gandiva and saw 70 X improvement in real-world performance and so that's not going to be every use case but it's a fairly powerful thing and that we've useful so that was Gandiva now there's a lot of other algorithms that you start to deal with once you're dealing with Aero memory I'm going to cover a couple of others so the first is is that when you're dealing with data one of the key things to do is feel to pipeline the system so if one process one process is waiting for another process to do work because you're doing a lot of work in one process before you hand data over then it's not a very efficient use of resources and so frequently we're very focused on pipelining and moving data quickly down the pipeline so that the next thing like shuffling it across the network or whatever can be done as quickly as possible and so generally speaking for us we generally keep these record batches 264 K or less in size the number of records that we have and then we actually size it down based on how wide the records are if you've got like a two-column record to column record well then you could probably have a pretty tall a record batch but if you have a very wide record with 500 columns you're probably going to go narrower and the reason is is that you want to generally speaking keep the arrow batches on your CPUs cache and you want to be able to pipeline them through the system as quickly as possible you do lose some efficiency when you're working on shorter vectors because you're gonna have to hop out and hop back in and work more often but the flipside is is that you usually can benefit from pipelining and keeping those things not too big when you're sending them places second thing so while columnar is really good at working with the cpu it actually is painful for some operations so one of the operations is kind of painful with a columnar is a filter so in a traditional processing where you're working with row-wise representations you'll just drop individual records as you don't need them right in a common representation you've got separate vectors for every value and so if you want to drop a value well then you'd have to recopy and compact all of those vectors and so rather than doing that immediately you actually apply a mask okay we call it a selection vector but you can think about it is just basically a set of pointers okay and we size the pointer based on how much data we're going to have and so we call so we use frequently what we call selection vector - which is a two byte pointer that points into a record batch which is why there's 64 K or less and tells us which records are valid in that record batch so that we can mask it so the additional operations that might be doing copies themselves like an aggregation or a join don't have we don't have to do an extra level of copying in between and then a selection vector for is basically two byte pointer into a particular batch and two bites into the which batch you're actually interacting with okay and we actually also have a six bite one that we use as well and so trying to stay as compact as possible but then solve those problems another interesting pattern that becomes a little bit challenging with Commodore data is a parallel or Scheffer let's say I want to shuffle data between ticket from one node and then hash partition it out to a bunch of other nodes okay normally you might do this again in a row wise format you'd apply the hash the hash function figure out the bucket and then send it on its merry way in a columnar format you're probably going to do it a little bit differently and we do it a little bit differently and so first we actually what we do is we MUX data so that we bring data together from multiple threads in the sending side so that we can kind of reduce the amount of buffering that we have to do before we send the data and then we actually generate a bucket vector which is a description of each value and where it should go and then we go column by column over the bucket vector and the specific vector and make a copy of that a targeting each of the outcome outgoing locations pre allocating all of the memory for those things and so one of the common things you'll see with all of the comment the columnar algorithms is a decomposition so that you're working on you may be reading the data more times but you're you're interacting with a smaller scope of data in each operation so once we have good processing which is what we find with with the Gandiva and these other algorithms we then want to move the data and so arrow flight is one of the newer components of arrow and it's this ability to move data between different systems it's focused on bulk transfers for analytics purposes and it's really sort of delivering on what we had as an idea four years ago which is full interoperability of data between different systems using arrow okay so really simple concept I want to send data between two different systems a client and server can send data to each other it can be started it can be initiated at either side and basically it says hey I'm gonna send some data here's the data okay I'm done right or hey give me some data okay here's the description of the data here's all the data and you're done okay so it's mostly a effectively a peer-to-peer kind of protocol but one of the key things it does is so we're focused on analytical systems we're focused on scale out and so one of the key problems with the traditional communication mechanisms is the typically single stream and so arrow has has at its core the concept of multiple streams and so you can go to a host that says hey here's all the different streams you can subscribe to for this data and it says here's here's a hundred different what we call tickets and so you need to consume all hundred tickets in order to get all of the data and each of these tickets can be accessed on a subset of the nodes that are available in your cluster and so you can actually go in parallel and have a hundred nodes pull data from another hundred nodes using a parallel flight stream so it's not just for a single stream although it works very very well for a single stream but it's actually also designed for parallel stream and so for example inside of dromio we now have a connector that allows you to consume dromio data in parallel into a spark pipeline so performance wise I've got a benchmark here so we did a very simple example which is accessing our tool dromio Jeremy exposes ODBC and JDBC and so we said hey let's go into a Python data frame we want to load some data from an ODBC driver okay we did that initially with PI ODBC which is what most people use and we pulled 5 million records back and in that situation it took about 54 seconds ok we then did the same thing someone I was I did the talk with just the top left it left and right side benchmarks and someone said hey turbo ODBC is way faster you really shouldn't be using PI ODBC as your benchmark so we ran turbo DBC it is way faster uses ODBC is the same thing they've removed a bunch of Python code and replaced it with C C++ code and it takes 22 seconds we then ran the exact same operation in PI arrow against the exact same system using the exact same stuff and we got it back in 1.3 seconds and so these times are all fine because I can sit there and wait in any of these cases when you ramp this up to a billion records it goes from something like I think it's two and a half hours to you know a few minutes something that you maybe go get a coffee and come back and so it can really change the the speed at which people can get to things so where are we going from here well the future is so we basically were building blocks up so we started with community we then introduced the format we then started adding language bindings we're continuing to add language bindings then we did file format integrations for things like CSV and park' then we introduced it the Gandiva kernel for execution then we introduced Aero flight for transport and we're continuing in that path of adding more things to it but most likely one of the systems you're using or a system that you're building should be using error so that's what I've got I put up some links here of different things so code examples it's hard to show them here so I just put them as links so you can go look at some of the things that was talking about in terms of algorithms feel free to try out drive me oh we've got an open source edition that you can download and play with and go to github and look at our code that's some of the links here again d even flight benchmarks that I referenced are also linked here so that thank you so much and I got like maybe one or two points for questions [Applause] [Music] [Applause] so it looks like you have a lot of the building blocks to replace spark and a variety of other processing frameworks you know if you had to say how you complement or could replace or you know how do you contrast with those types of frameworks versus addres mio and its capabilities yeah so well I think it'd break into different pieces because there's arrow there's drum you right and arrows open-source and so the hope is is that technologies and that's already happened in many cases and spark adopt arrow as a kernel to solve different things so dramatically high concurrency bi and that's kind of our sweet spot so if you're trying to do a lot of analytics I'm trying to solve how you get to data we saw more of that where as we see people doing more data science workloads and ETL workings on something like spark so if you were going to compare us to something you'd probably compare us more to something like presto or snowflake another question in the middle here thank you so one thing that I've sort of gathered over the past few years is one of the reasons why MapReduce sort of stalled is that it turns out that data movement is cheaper than people expected like so there's the it's okay to just to just spend 40 gigabits on like moving data between between tasks you mentioned already it may very briefly do you find that in your experience in sort of production customers and larger deployments are people actually like do they really need to shift the data or is it okay like is the real adoption of like leaving the data in memory at you know the place that has custody of it and just accessing it remotely yeah it's interesting there's a couple different pieces in there so I absolutely have the same observation I think you have which is is that so the big rise of Hadoop and the passion there was all around push your processing to your data right and that made a huge amount of sense when google wrote the original paper and everyone was on a 10 a 10 megabit in a network connection now that everybody has 10 gigabit 40 gigabit 100 gigabit connections it's not as much of an issue especially since mm I don't know the vast majority of people are moving their their large data links to the cloud and there is no ability or there's very little ability to be able to push any processing do that and so generally speaking I believe that people should stop worrying about pushing their data they're processing to the data and just focus on solving their problems now aro still fits very well in that because our fashion is not necessarily to hold the data in memory all of the time right the question is how do we optimize so it give you sort of a sort of a very short interesting thing which is is that everybody who Creed's park' in a production environment whether it's spark or presto or addres mio or Impala all those tools actually built their own park' readers and the reason that they did is because they had the same problem of basically reading and sucking everything through a straw of an API that wasn't that efficient for how they wanted to internally represented the data ok and so what's happening with aro is is that you now have a representation where people can pick how they want to write their API so that they are just focused on the data format and so what we're starting to see is a consolidation of these things back in to a single version of that code that's been written and then different people can decide how to interact with the data once in the scenario format and so give you an example of that as I mentioned earlier we're very passionate advice but one of the things the iceberg community is doing is exciting their system to be exposed as a narrow representation so if any system that understands Aero can consume them one thing that I've been working on a little bit the past few years is is stateful streaming and I'm wondering if Gandiva has any support for state either within batches or between batches right now again David doesn't have any support for state it's a design to be an open framework so people could add it but it's not right now it's a stateless system up beyond the state of the expression itself alright thanks everybody I'm happy to ask more questions [Applause] [Music]