SBTB 2019: Michael Paul Armbrust, Delta Lake: Open Source Reliability and Quality for Data Lakes
[Music] Thanks so I'm super excited to be here today to talk to you about this new technology that we've built called Delta Lake which as he just said helps you bring data reliability to your existing data leaks but before I jump into that first of all who's familiar with this term data Lake is this something everyone knows okay most of the audience I want to tell you a little bit about why people are excited about data leaks and why people have been spending millions of dollars setting up data legs and trying to get insight from them so the idea is this you have a lot of data and it comes in a wide variety of formats it might be clickstream data it might be metrics it might be console logs it might be structured sequel data it might even be unstructured things like video and speech and the magic of a data Lake is you can store it all in one place it's not a traditional database you don't have to start by doing an expensive ETL process where you decide exactly what the right schema is it's just a file system so you can just dump everything there and you can keep everything and this is actually good because it you know it cost something to spend all that time trading the schema and you don't know what data is valuable out priori sometimes it's helpful to just keep everything and then look through it later and the idea is once you've collected everything in theory you should be able to do data science and machine learning so you should be able to build recommendation engines will help people find the things they're looking for should be able to prevent fraud and decrease risk or you could even do crazy things like cure cancer using advanced genomics but unfortunately I have some bad news for you it turns out that most likely your data is garbage and if you know as a result the data that you've stored is garbage and you have this kind of standard problem of garbage in garbage out your analyses are garbage as well and why is that true why is it so hard to get high-quality data inside the big lake I want to walk through kind of a story that I've seen happen over and over and over again as people build together systems where they try to bring reliability to these data leaks so this is kind of a pretty standard problem my boss came to me and said I've got a bunch of events they're coming in through Kafka but it could be canoes or s3 your event table or any other system and I have two missions I want to do streaming analytics so I can understand what's happening in my business in real time right now and I also want to do AI in reporting where I take a more longitudinal look and I actually look at trends and make predictions about the future so how are we gonna do this well I might be a little biased but I would start by using Apache spark we have streaming API is that allow you to read from sources like Kafka you can do event time aggregation and all the other kinds of cool things you want to do and you can get your streaming analytics great we're done right well unfortunately challenge number one is historical queries things like Kafka are great for providing a low latency buffer to bring data into your system they're not great at storing all of the raw data for years and years and years of analysis and so I read a lot of blog posts I did a lot of thinking and I came to this thing called the land architecture which as far as I can tell means we're just gonna do everything twice we'll have a streaming system that tells us what happened what's happening now we'll have this separate batch system that will archive everything and then you know once we set that up it's one more thing to add to our architecture but you know spark has unified API for streaming and bat so it's hopefully not too much work and then we can do our AI and reporting great well unfortunately this brings us to challenge number two which is messy data as I mentioned before most data is garbage someone upstream from you decide to change the format of the date without telling you and now you're parsing is wrong and so we got to like figure all of that out and so I've seen a lot of people do this and what they usually end up doing is they write a whole bunch more SPARC programs and those are gonna do validation they're gonna check to make sure that the data that I expect is actually there and if not they're gonna fire off some alarm so I can go and fix it and you know since we lamb to architecture I have to do this in two places I have to do it in both streaming and batch but unified API is make it not too hard so I can do that we can add validations to the architecture and we're good to go right well unfortunately this bring us to mistakes and failures I forgot to put a validation in some bad data crept into my data lake and that's difficult to fix because as we mentioned before this isn't a database it's just a file system so going in and modifying the file system in places all kinds of you know concerns can I even find the data that I need to change happens many crash in the middle am I gonna corrupt my data and so I've seen another pattern here which I bet many of you have actually probably implemented yourselves where you're gonna take your data lake and you're gonna partition it you'll break it up by day hour month minute whatever makes sense for your scale of data and you'll build an engine that allows you to recompute individual partitions so when something goes wrong rather than reason about how to fix it I'll just delete that day and recompute that day from scratch and this is great because now I can fix those and the mistakes that came into my my data lake this is you know one more thing to add to our complicated architecture but with it we can now have a little bit of certainty about what's actually going on we can fix those mistakes that unfortunately brings us to problem number four which is updates you know reprocessing is great when you have one partition that needs to change but what happens when something like gdpr happens you actually need to scan through your entire data leak and allow somebody to be forgotten or if you wanted to change data captures we talked about in the earlier presentation I need to modify you know records all over the data lake and you know very quickly and and that's very difficult so you can build a whole nother set of spark jobs that handle these updates and these merges but you got to be very careful here because if you modify the data lake well somebody's querying it they're gonna get the wrong answer they're gonna see partial results their analysis will be wrong so you need to be very careful to schedule these at different times I'm gonna run my gdpr once a month at 3:00 in the morning it better finished by the time the batch jobs start running or my reports are gonna be late but it can be done we can add that to our contexture and now we have updates but you're probably seeing a theme here and to me the problem is we're wasting a lot of time and money solving these well-known systems problems rather than doing what we really want to do which is extract value from our data and the distractions that are getting in your way and preventing you from getting that value out are a couple of different things I think a big one here is the lack of atomicity Animus t means something either happens completely or not at all and when you don't have it anytime something crashes in the middle which will happen in a distributed system you have to reason about what happened and what didn't how do I clean up from that failed state and that's a lot of bug you know bug-ridden tedious code to write there's no quality enforcement you know nothing prevents anybody from dropping an extra file into that directory or not it matches the schema that you're looking for and that can make it very difficult to come back and look at it later and then finally there's no power for consistency or isolation which makes it very difficult to do things like mix streaming and batch on the same data set or query a data set while it's being updated and these are all things that would be really nice they were drastically simplify our architecture if we had these properties so let's try it instead with Delta link the idea here is we're going to take this relatively complicated architecture and we're gonna switch it to something like this where rather than thinking about all these systems problems we're only thinking about the flow of data through the system and why are we able to do this well I think one of the most powerful kind of techniques that we're gonna bring to this this problem is full asset transactions so this gives us a de Missa tee consistency isolation and durability and this means that you can focus on your data flow rather than worrying about these intermittent failures Delta Lake is of course open source and open based on open standards the data is encoded as park' it's released under a full apache license with no weird commons clause restrictions where you can't use it if I think you compete with me and we actually recently joined the Linux Foundation to kind of create a permanent vendor-neutral home for this so you can store petabytes of data without lock-in which i think is really important data has gravity data has inertia if you're gonna collect this stuff that is so value to your business you don't want it to be in some black box format that it's gonna require an expensive ETL process to get it out if you ever decide to change things you want to be able to query it with multiple engines and as I'll talk about we're working on you know adding support to other engines as well but today spark Delta Lake is deeply powered by spark which gives you the ability to work in both streaming and batch and if you have existing spark jobs as I'll show in a couple slides it's easy to convert your existing jobs with minimal modifications so now let's simplify the picture and I want to talk a little bit about some patterns that emerge once people stop thinking about these systems problems and start thinking about data the first thing that I notice is people start thinking about incrementally improving the quality of the data until it's ready for consumption and why is that why do you think about these things incrementally well I see a couple of different classes of data they'd like to talk about so these are kind of high-level concepts these are not baked into Delta you do not have to have a single bronze Ingold table but I think these are useful ways to communicate with each other about the quality of your data sets so starting off with bronze we have our dumping ground for raw data you often keep years of data here and you might say wait a second why am i storing this raw data well there's a couple of reasons for that and I think the most interesting one is there are no bugs in a parser that you don't write if you just store the raw data you can always go back to it and reprocess it from the beginning of time and Delta is designed to store years and years of data in a single place moving on from there we have our silver tables these are cleaned up a little bit you may be joined filtered aggregated them you've taken some JSON data and parsed it out to top-level columns this is intermediate datasets where you've done some cleaning but it's not quite ready for consumption and you might say wait a second why am I wasting the time materializing these intermediate datasets I think there's actually a couple of interesting reasons number one is you're probably not the only person in your organization who will find value from this partially clean data and by materializing it as a table this is a great point to fork off and allow everyone to kind of benefit from the cleaning that you've done but another reason is which I actually was kind of surprised by debugging when you get to your final analysis and something is wrong having that intermediate data set is a fully materialized table that can be queried with the full power of sequel means that you can actually figure things out about where things started to go wrong you can check and see wait how many times is this column Knoller how many distinct values are here in this intermediate step and that makes it dramatically easier to understand what's going on and then finally we have gold data these are high level business aggregates that mean something to somebody important you can power your reports you can do a I on it this is clean data that's ready for consumption and you can read this with spark and you know we're kind of extending this to work with the Presto community as well we're currently working with starbursts on that another pattern I see is a lot of people switch to you streaming and I want to talk a little bit about this because I think streaming there's some misconceptions about what it's for a lot of people think that streaming is an extra complex thing that you do when you need high / low latency things for for some reason and to me that's that's one thing you can do is streaming but that's not really that's missing the power streaming to me is about incremental computation it's this problem where I have this transformation that I want to do and my data set is continually changing that's pretty much everything we do with data nobody has a static data set and what the streaming engine does is it allows you to kind of stop thinking about these problems of what data is new what data is old how do I incrementally process the new data and move it downstream how do i transactionally commit it guaranteeing exactly one semantics how do i checkpoint where I'm at so I can recover from failures streaming takes care of all of this for you and there is in SPARC a very nice kind of cost latency trade-off that you can kind of make automatically if you do really care about hyper low latency millisecond processing you can run in continuous mode and we will hold on to a core and continually pull Kafka and give you hyper low latencies if you care about seconds to minutes of latency you can use micro batch processing where we will run many different streams on the same cluster each taking tiny little head box of time on each of your executors and finally if you have a data set where data only arrives once a week don't keep that cluster running all the time use trigger once and trigger once the system will boot up process everything that is there and then if you're running in the cloud shut down it's an amazing way to take advantage of the elasticity of the cloud while still getting the benefits of streaming you can save a lot of money by turning your cluster off when you're not using it now of course streaming is great but batch jobs happen and there are a lot of reasons why you might need to run batch jobs as well so I mentioned gdpr once every month you might want to sweep your data leg for anyone who sent you a DSR you might have retention requirements where you only want to keep you know the last two years of data and then remove it or you might kind of want to do up search so you could do change data capture from some operational data store and Delta Lake supports all of these on the same data sets that you're streaming to in from another pattern that although I all noticed recomputation if you follow all of this stuff it's very easy to when you have a new analysis that you want to do or when you notice a mistake in your existing analyses recompute from the beginning of time so the reason that this is so easy is because a stream in Apache spark actually computes the same answer as a batch job over the same dataset so when you start a new stream what it's going to do is it's going to take a snapshot of the table at that moment break that up into pieces process them incrementally and then when it gets to the end of that snapshot you will start tailing the transaction log and incrementally processing new things that have arrived since the stream started so when you find a mistake it's as easy as clearing out that table is creating a new checkpoint and running the stream from the beginning of time and it'll automatically kind of scale out and use a whole bunch of machines early on and then when you get down to that incremental processing you can scale that cluster down to something significantly smaller so next I want to talk about who's actually using this and you know before we released this as an open source project Delta was a kind of for the last two years proprietary product inside of data bricks it's used by thousands of customers around the world and I actually need to update this slide we processed over two exabytes of data last month I want to talk about one particular use case though that I found pretty exciting this is Comcast they use it for session ization so they have this problem where they're you know millions of people around the United States clicking with their remotes going through from the Home Shopping Network HGTV back to the Home Shopping Network and they want to figure out what is that content journey like they want to understand how people are consuming their content they've been doing this with Apache spark for years but they have this problem there's too many subscribers and so the scheduler would just tip over if they tried to run this on a single cluster and so you know being good engineers we know how to solve scale problems will use hash partitioning so rather so they had to do was they had to take this job split it into ten different jobs running on ten different clusters with ten different schedules ten different sets of logs ten different sets of errors to handle and manage but they were able to get it to work however when they converted this to Delta because of our scalable metadata handling and because of kind of our efficient use of transactions they were able to collapse us down into one single job and that actually has huge implications for their productivity when you are managing one thing that's much easier than managing ten and they also reduce the number of machines that they needed by 10x as well which is a dramatic reduction in cost so I have all of this extra money to do other cool different analyses rather than wasting it all on this one so how do you use it it's pretty easy to get started it's published out on maven if you're already using spark all you need to do is add this one command line argument it'll automatically download and install Delta on your if you're using maven or SVT the coordinates are up there you can add it to your project and in terms of code it's actually also very easy if you're using the data frame reader and writer all you need to change is from park' or CSV or JSON whatever you're using today to Delta and everything will work as it did before except now you have asset transactions I want to give you just a preview of some stuff that we're working on as well this is this is kind of a project that's underway that we're hoping to open source in the in the next couple of months but the idea here is transactions are great for making sure that your code runs correctly but it doesn't prevent you from having bugs in your code in order to do that we need something a little bit more and the idea here of Delta pipelines is we're gonna give you the ability to declaratively specify your entire production ETL graph as one single entity and this actually makes it very powerful so looking here I'm creating this thing called a dataset I'm using the kind of familiar API so I already know and love from Apache spark just data frames data sets Scala Java Python or even just pure sequel I can declaratively specify you know where it should be stored what requirements there are on the schema whether or not it should be registered in a meta store so that so that other people can discover it and I think this is actually also very powerful I want to talk a little bit about the schema in particular Delta has kind of tunable ability to make this the schema very strict in kind of your bronze tables I will often see people use automatic schema evolution so Delta will just automatically add any new columns assuming there aren't conflicting data types as they arrive but you can also act more like a traditional relation relational database and say nope anytime the schema changes I want somebody to run an alter table and do strict DDL it's more common in the Bruns or in the gold data sets but my favorite feature is this thing called expectations expectations allow you to take your domain knowledge about what correctness means and actually put them into the system for those of you coming from a traditional relational database that sounds a lot like invariants but with a key difference and invariant will kind of fail the transaction anytime something goes wrong and with expectations we can actually have tunable severity so you know in this case I'm defining this expectation that there's a valid timestamp and it's actually more than the timestamp just being present I know that data break started in 2012 and so if a record appears from say 1970 that is probably a missed parse and not actually a valid date and so I'm kind of putting that domain knowledge into the system and then I can choose what to do when that expectation is violated if this is a final table that I you know report to the IRS I probably do want to fail that transaction make sure no bad data ever makes it in there however if it's an earlier table perhaps I just want to monitor the quality so I want to alert if it goes above 5% or my favorite is a quarantine so any time an unexpected record is seen rather than shut everything down we allow processing to continue and we take that unexpected record and route it to a quarantine table where an engineer can come and look at it later so this allows you to continue data processing your data while still understanding what's actually going on under the covers I've been using this on my own data sets and it's amazing what you find because in many cases your expectations are wrong as we talked about before your data is garbage and we tied deeply into Apache spark so we'll actually modify the query plans we can say things like here was the input record here was the output record here was the expectation that was violated here was the code that produced that invalid record it makes it much easier to understand what's going on under the covers so finally I want to talk a little bit about how this actually works under the covers I want to kind of deep dive into the nitty-gritty details because asset transactions in a distributed system sound too good to be true in some cases so starting off with how you know the kind of file format for Delta don't on disk will look very familiar to any of you who already have a data Lake is just a collection of park' files and a bunch of directories but we add to this a transaction log and the transaction log stores different versions of the table so here we have version 0 and version 1 of the table alongside these kind of optional partition directories and data files now a version is defined as the set of changes from the previous version so what kinds of changes are we talking about well you could change the metadata so if you for example add a column to the table or change the partitioning of a table you can add data to that so if we do this at the file granularity so you can add a file along with optional statistics like min and Max values for each column that allow us to do really cool tricks like data skipping and then finally you can remove data from the table so you can remove a file from the table and if you think about this you can now take the transaction log from the beginning of time and play it and when you get to a specific version what you will have is the current metadata for the table and a list of the files that are currently in the table so we're using a trick here called multi-version concurrency control many copies of your data live together in the same directory but this is the authoritative source for which files are valid at any given version now we want to get those really nice acid properties on top of this so how do we get out of necessity well we're gonna basically take each of these individual units these kind of Delta files and we're going to use the atomic primitives of the underlying file system to create them so if you're storing this on s3 puts to s3 are atomic by definition the way they get this is when you start the put you say expect this many bytes and unless they get that many bytes the put doesn't go through so you can add a Mississippi kind of out of the box on other file systems like is your data Lake or HDFS we'll just use atomic rename so we'll create a temporary file and then move it to its final destination that way you'll never get a partial file so in this case you know we've added two files and let's say we want to atomically compact that we want to take them and collapse them down into a single file well we can do that if we didn't have a atomicity here we'd be in a really bad space because if we deleted those files without creating the third we would have lost data if we added the third without deleting them we would have duplicated data either of those would be bad for our data quality and so this add Amissah T is actually very powerful the next thing we want is serializability we want everybody to agree on the order of changes that we can kind of have a consistent view of the data at all times even though there's multiple people modifying it so here I want a property called mutual exclusion if one user creates version 0 great if another user creates version 1 that's awesome but if two users try to create version 2 at the same time one of them needs to fail and again you know on s3 this is actually impossible it is not a lock service so you can have kind of another database that mediates who Weiner is that does give you this property on things like HDFS and is your data Lake there rename operation actually can fail if the destination already exist since you kind of cut this property out of the box now you might be saying wait a second if every time two people modify the table at the same time it crashes well that's not very great for me as a user right I thought this was going to make my problems easier well fortunately we have one more trick up our sleeve which is called optimistic concurrency control so we're going to optimistically do these as though nobody else is modifying it and we're going to check for conflicts among these operations and in many cases there will be no conflicts and we can move on the algorithm is as follows so let's say there's two users who are both streaming into the same table so they record the start version so they say okay I am streaming in starting at version zero of the table they record what they read from the table and also what they write to the table so in this case the only thing they need to read from the table is the schema they need to validate that the data that they're writing in matches the schema of the table and then they're both going to attempt to commit so they'll say okay I wrote out all these files and I want to add them to the table in this case user one wins and user two loses and so what what user two is gonna do is they're gonna check and see hey did anything that I care about change and in this case the only thing they saw was the schema the schema did not change and so we can just lie to them and say actually you came afterwards and they don't know the difference and so we kind of automatically run this transaction again nobody notices that there was a conflict things continue on automatically we have one final trick up our sleeve which is in a data Lake you can get really massive collections of data I I've seen a couple that have you know millions to tens of millions of files in them and so at this point the metadata of the table actually starts to become a big data problem in itself how are we gonna solve that well fortunately we have a big data system spark so what we're gonna do is we're going to design the transaction log in such a way that it also can be processed by spark so we'll take all of these actions will load them into spark and we'll do a shuffle will resolve them so we have only the current set of files and then we'll create a check point which we actually also stores park' this nice efficient columnar format so now when you want to query the table rather than replay the log from the beginning time or make a lot of really slow metadata calls to s3 to figure out what files are there you can actually just go directly to this checkpoint and you can even do things like say you know this table may have tens of millions of files in it but I only care about yesterday so you run a really quick distributed query to identify the files that are relevant to yesterday and then you'll run another query over the file set that is returned and this allows you to scale much further you know then you would if you were storing this in something like my sequel or the height meta store make anybody who's tried to put tens of thousands of partitions and the height meta stories kind of felt that pain so finally I want to talk a little bit about wrote the roadmap for the project you know as I said before this is a relatively young open-source project but my goal is to have full API parity by the end of the year so that has actually been the last couple of quarters of work we added support for s3 and Azure blob store we added the ability to do all these cool DML operations update delete merge you know we added the ability to convert existing park' data into Delta and do describe history and with sparkler point-o will have kind of full support for DDL the create table alter table and all of those kinds of things so with that I'd like to encourage you to head over to our website Delta dot IO join the slack channel we hang out there all the time we would love to answer questions during the mailing list for cus on github yeah thank you very much great any questions for Michael so can you discuss the challenges of implementing additional data file data file formats you mentioned parkade it isn't easy to just also support Avro yeah that that is a great question so if you and so we actually just published the protocol spec up on github and there's a section on that we actually already support any file format that spark supports now there is some interesting it is not exposed to users and I would love to talk to more people about this because we're very open to changing this but when you're designing a system or a product I like to kind of think about how can I remove tuning knobs and so we picked Parque because we know it's fast with spark we know it's kind of pretty optimized there and I've actually had a couple of customers where they took their existing data that they were storing in something like hi of sequence input format and they converted it to Delta they did no other tricks they didn't do compaction they weren't taking advantage of asset transactions nothing else but their queries went from taking hours to taking minutes and they were like oh man adults is awesome and I was like actually I just forced you to choose a reasonable file format now you asked about Avro I think there actually are like interesting differences between parking and Avro wherever it might make sense so yeah please go to github tell us what you're trying to do and I'd like more than happy to kind of open up extra extra knobs you know if they're useful any other questions yes yeah absolutely so you know just for those of you don't know iceberg is an Apache project actually no Ryan blew pretty well he's the one of the creators of it I think you know in design iceberg and Delta are actually pretty similar so you know I think if you're going to try to build asset transactions on top of cloud storage systems the design most people have come to is you will have a separate place for tracking metadata at the file level rather than at the partition level we kind of all have this so I think you know in that level they're they're very similar I think where they differ iceberg has a couple of features that I'm like really excited to copy in particular I think computer partitions are really interesting and also they have a slightly better ability to rename columns through a lot extra level of indirection they have now the difference with Delta though is you know I think Delta has full support for streaming which I think is critical to running data pipelines at scale we have update delete and merge and I think the kind of primary difference to me as Delta isn't used by thousands of organizations around the world in production and has been for two years we store multiple petabytes of data we process exabytes per month iceberg is a relatively young project any more questions hi you mentioned that uh when you trying to consume from death lake you can enable it for the outer layer latency ultra-low latency by enable you know pulling from Kafka all the time so that's that's the spark streaming streaming really guarantee that is the per message processing oh it's his central messages procession let's similar with the Kafka stream essentially it's a steel a micro patch yeah great question okay so first let me clarify I'm an answer for put my apache spark hat on and then I'll answer for Delta Lake next so yes spark continuous processing is true streaming it sits there holds on to the core polls from kafka supports millisecond latency do not listen to the Fudd that says spark is only micro batch that has not been true for two years now Delta whereas Delta sweet spot Delta is about creating bulky park' files that you store in a system like s3 we're not talking milliseconds anymore round trips to ask three or like 500 milliseconds already the sweet spot for Delta is seconds to minutes the color steady short like these bronze the gold silver so yeah that's a great question bronze silver gold is that part of Delta know what I here's what that is it's a common misconception so there is nothing in the Delta API that enforces bronze silver and gold but that is a pattern that I notice that people get into once they stop thinking about systems problems and start thinking about data quality instead so that is just a very useful vocabulary that I have seen evolve you can not use it you can have ten bronze tables and skip directly to gold it's all up to you and with that a query time thank you very much that's all we have time for thank you very much Michael appreciate it next talk begins in 30 seconds