Making Apache Spark Better with Delta Lake

SF Scala: Chris Hoshino-Fish, Making Apache Spark Better with Delta Lake

Recording: SF Scala: Chris Hoshino-Fish, Making Apache Spark Better with Delta Lake

[Music] thanks again Alexi thank you all for coming as Lexi said my name's Chris fish I'm solutions architects at data bricks and before that I was a data engineer working primarily with spark been working with spark for about five and a half years since version 0.8 as most of you know Delta and Delta in particulars what we're going to talk about today spark as well right as a scalability project implemented in Scala and then a lot of people also use it in their AI projects for big data distributed computation things like hyper parameter tuning that kind of stuff so we're gonna focus primarily on Delta Lake today Delta Lake the other person's name here Michael armbrister of the Corps spark committers and he's the creator of spark sequel as well so he's got a PhD in database systems and then he implemented spark sequel for spark so he's the author of data frames data sets big on streaming as well and then Delta Lake arose out of not just the challenges he experienced managing his own pipelines but what we saw at all her fire customers as well so we like to start off by sort of telling the story that motivated Delta in the first place and it really starts with the promise of a data Lake right so businesses have data being produced by things right and in today's world we have data being produced by a lot of things you may have customer data coming from your iPhone devices you may have data coming from sensors that you've placed somewhere might on a website that's producing a lot of data and we've realized over time that data is actually very valuable and storage has become very cheap right so we want to capture all of it and we want to put it somewhere that then we can query but we don't want storage directly in a database because that will become expensive and our Oracle bills will go through the roof so instead we're going to store it in a more unstructured way in data Lake and that's what most people do right and then okay great so the business says why are you doing this well we're gonna be able to do all this great data science and machine learning with our data we're gonna build recommendation engines we're gonna build predictive engines we're gonna evaluate our risk better we're gonna have greater insights or maybe we'll do things that have never been done before like we'll be able the sequence full human genomes and do analysis on that right so the promise of data is great but what really happens in reality is your systems are maybe producing terrible data they're producing inaccurate data they're producing wrong data they're structuring it and properly and then now you've stored all of that but it was garbage in the first place and now you've just stored your garbage in a garbage collector and then your predictive systems are also terrible because the data quality going in wasn't good in the first place especially you know as you get more and more into data science or machine learning you learned that it really is about the quality of the training data and then not just the quantity at all and so this is actually a huge problem because you can't derive value out of something that you can't have certainty about so how does a typical data Lake project actually evolve a lot of people will start by collecting events right and usually you'll collect them into some sort of system like Kafka and eventually start to retrieve them right so I have data flowing into my append-only event log and let's say okay the CEO says I want to know real time how many people are using the app okay so you say all right I need to build streaming analytics so I need to build a stream I need to write my code I need to point it directly at Kafka and pull the data right away and so now I have a streaming report going on but how can I do historical queries on this data as well right I can't do historical queries in my stream my stream is focused on the real-time live data instead I need to possibly do something like a lambda architecture so maybe if I want to do historical queries I'm gonna have to weigh two things pulling from Kafka one is creating a real-time report but the other is just syncing my data into data Lake where I can go back and look at it offline and so okay this is working pretty fine you know maybe I had to configure my Kafka topics a little bit differently load-balanced a little bit but for the most part things are fine I have two streams pointing at them one's going to a real-time report the other is going to my de Lake but this was a little bit of implementation overhead I had to tune some things I had to go in I had to implement the code okay now so now I can query my data Lake I can go in and I can start building out you know a data scientist says oh I actually want the data eight aggregated in this particular way because that's the feature that has the most predictive usefulness for the model I'm trying to build so you go in and you build some nice pipeline and you pull them the business data they want and you start building those things but then the date scientist says hey the data you've put here in the data Lake does not match up with the report you've been showing the CEO and it turns out I'm saying something different to the CEO and he's been telling the board these other numbers which ones are actually accurate here so you say okay well I am NOT these are two separate processes so I need to do some kind of validation between I need to look at both stream and offline reporting I need to check do these numbers actually match up okay so that's not you know crazy difficult to do I can do that right I can I can take wherever I'm putting the streaming report to potentially you know some database I can pull those numbers and then offline I can run a job and compare them to the data Lake I say okay okay so I can validate these numbers definitely I can compare it can make sure that they're all accurate but again I did have to do some implementation right I had to start pulling this data maybe duplicate some data again and I had to set up another job in order to validate that these were actually accurate but let's say my validation job actually does find an error it finds a discrepancy and it says oh you know like over here in the data Lake we actually messed up last month's data and we want to rebuild those metrics so how am I going to do that right sparks native data sources don't provide me away too go back and transactionally update data in the past right so what a lot of people do and what I have done in the past and pray a lot of us here is you partition your data lake by date and then you can go in and you can simply delete the old data and put the updated data in the same place with the date and so this lets me do updates like I'm having to delete and rewrite a lot of data in just in order to do updates but I am able to do some kind of updates if there's a problem again now the complexity of my project has grown grown quite a bit but let's say I want to go in and do just a fine-grained update like for instance I got a gdpr request to delete someone's data or the newly released CCPA which I'm sure caused a lot of roadmaps to get in this step well that's really hard to do actually because I can only over right at the partition level with park' or orosi and so even if I just want to lead a single row in my data Lake I actually need to delete a whole folder and rewrite everything and so okay I can do this yes but this is now starting to add a lot of operational complexity over all right and especially if something breaks in the middle if my right job fails in the middle maybe I'll lose that data forever hopefully not but even if I haven't I may have left partially written or corrupt data there in which case I'll have to go in and clean that up manually and then reschedule all of my downstream jobs and play the scheduling game to make sure things are running in the correct order and the data is being picked up by the downstream jobs that's the next slide so okay but I did all of this that like I implement all of it it works it does the correct things I'm confident in my metrics and confident in the numbers but what if I spent and what costs have I incurred it doing all this I've wasted not wasted but I've spent a lot of my employers time and money doing they've spent a lot of my time doing this I've also caused a lot of headaches for myself had to learn a lot of things I've had to debug love problems and the main thing is that I'm not focused on the data right I haven't really spent a lot of time learning about my actual data instead I've spent all this time learning about the storage systems about the query execution engine about bugs that could exist in there but I'm not learning deeper insights about my actual data which I actually am trying to drive business value out and that's what you really started this project for right was to try and derive business value out of your data try and accomplish something with it and instead you ended up doing all these other things which are fun and interesting but not the original purpose of my project so these are sort of the the main distractions that are causing these problems right Adam is me atomicity right for the a and acid transactions means that when my upstream job fails it leaves behind partial data or corrupt data and it breaks downstream readers my quality enforcement right in a normal database we can set strict schema enforcement rules we can say this column is always an integer this column is always a website this column is always an email address I can't do that in my data Lake instead I need to write my own code to do that and implement it in a pretty heavy way I need to go in and make sure every job is enforcing the quality that itself and then there's no consistency or isolation between reads and writes and so this makes it really really difficult to mix streaming and batch workloads here so it's very hard to stream and batch right into a single data source it's also very hard to be streaming and batch reading off of that data source at the same time and really a lot of the scheduling game is just making sure a certain table is actually available to be read at the time I go to read it so how does Delta Lake actually improve upon this what features does it offer and a recent announcement for Delta Lake as well as that was donated to the Linux Foundation in November so this is sort of like moving it out of data bricks you know corporate ownership and putting it in a place where it can actually become a full-fledged open-source project and one of the main things here that I do want to try to get from people is the project itself has a lot of room to implement a lot of things a lot of like traditional database techniques over the next few years will be built in it so if you're an enterprising young software engineer or really experienced software engineer as well and can actually add a lot of knowledge to the project it's a really great time to get involved in Delta Lake so instead of this mess of having to schedule things in repartition and do validation and updates and merges are difficult instead we want to think more just about how the data is flowing and to end right we have on the right hand side we have what we want to do with our data we want to get real-time analytics we want to do artificial intelligence we want build models build feature engines want to do reporting we want these things to be accurate and we want them to be running all the time we're on up-to-date data on the left hand side we have our raw data sources our kafka Kinesis event hub in measure or text-based sources as well and all we want is to set up the flow we want to focus on our business logic and not focus on systems problems and have everything just naturally flow through so Delta provides full asset transaction compliance what this means is that any writer that's writing to the table either succeeds or fails and never leaves the table in a corrupt state so any writer will never affect a downstream reader because of a snapshot isolation so when a reader goes to read the table it'll read actually a specific version of that table and so an upstream writer fyz it won't actually affect these downstream jobs on the flip side on the upstream side you can be streaming into a table and at the same time you can batch right to it on the sort of on the back end so let's say you want to like fix last month's data but you have live data streaming in you can do that because again because of snapshot isolation so just like a database you get these acid transaction guarantees now and again one of the main reasons we decided to donate it as an open source project is because storing data is a big decision right data itself is hard to move around if I have 10 petabytes of data I'm not likely to want to store it in some system that is and I have no insight into right and I don't want to be locked into that system either so I'm going to if I'm going to choose to put that much data into a particular storage system I won't be able to understand how that storage actually works and then most people in today's tech world have standardized on using Apache spark for ETL for doing most of the data writing and in some cases you know of their engines length link as well so Delta's designed first and foremost for writing to by spark with both the streaming and batch use cases in mind and so this lets you convert your existing jobs with minimal modifications it's literally just a new data source for spark instead of doing format park' you're doing format Delta and we'll see an example of how the actual syntax works later on and then under the hood we'll get into this later but under the hood Delta is actually storing the data as park' and then doing the metadata management for you so in the the architecture pattern that ends up arising as a result of this is that we see people dumping very broad data into bronze-level staging table so your Kafka has retention period you know of a day or a few days so you need to get that data out of Kafka same with Kinesis and you want put it into cheap object storage so you immediately sync all of that raw data into a delta lake but you do very minimal processing here so use we have some customers who will leave the data as binary data as byte arrays or other customers who will do minimal JSON parsing and then put the data into a bronze table then you have these middle mid-level tables where you maybe you've started to do some small aggregations or you start to pull in business logic you're doing customer lookups or other things like that and then this data is you know maybe consumable by some of the you know higher-level people who day scientists who want to go further than just the highest level aggregates and then on the far end you have your business reporting level aggregates that you might point something like tableau at or some other reporting feature you might show this to your executives and then the bronze here is still on fire because that data is hot this is where your live real-time data is going to be coming in and you're not going to be altering it very much right so this data is not that valuable yet but then the silver data is very valuable already because you started to implement some of your actual business logic you start to pull in extra information and this is where you're going to go if you find issues with your top-level aggregates right you're gonna go to the underlying data set and say okay like is there a problem with the lookups and my exploding the data somewhere might double counting something somewhere and then you can also point other groups to this like de scientists will often time prefer the silver tables over the gold tables and then over in the gold tables really the highest level data that you can expose to your analyst large groups of people who expect numbers to be set in stone and expect them to be accurate and in particular we just released a native hive reader for Delta and then currently you can query Delta tables from presto but they don't always get the latest snapshot of the table so we're working with starbursts at the moment to build out the presto connector as well and then the other thing is that because it's inherently designed as a streaming data source you can actually start moving a lot of workloads over to a streaming abstraction and this is really powerful for a bunch of reasons a lot of people think that streaming means like I'm doing something real-time I'm doing a real-time use case and I need you know sub-second superfast processing in reality streaming has the same API for data frames as batch workloads in SPARC right so your code is easily swappable between the two and then streaming adds a bunch of stability improvements to your batch workload so streams process the data incrementally so over time as you add new data you don't have to reprocess the entire batch every time instead you can just update with the latest data this can minimize your compute costs as well as minimize sort of the actual access costs that s3 or blob storage will charge you right and then if you need to you can have low latency jobs you can move over between swap between like a time-based trigger or a manual trigger and then also this sort of eliminates having to schedule batch jobs as well you can leave these streams always running and expect that the data will end up where you want it to without having to consciously think about scheduling these things and especially you know like you can automatically retry so even if the streaming jobs do break you can easily restart them without having to do manual intervention and then because the creator of Delta is the SPARC sequel person it also supports all of your traditional database sequel syntax so it has support for full merge syntax inserts updates overwrites but in particular full merge syntax so when matched when not matched do arbitrary commands and then it executes these merges into park' and has a bunch of s3 or blob storage optimizations built in as well so it's make sure it really simple to implement something like a gdpr pipeline where you have to delete people's data I've personally been thinking about sending a bunch of CCPA requests to whatever companies on my data at the moment so expect some from me and then when you do have a problem it's very easy to restart especially with streaming with streaming you can just clear out the checkpoint and the downstream table and then restart your upstream stream job and it'll just start from the beginning and go until everything is complete and so I don't know all of us who manage pipelines have this happen from time to time you mess up business logic and suddenly you know a swath of data is now incorrect in some way this makes it super easy to go back and reprocess those things so this is just sort of like a mention of the the thing is we open-source Delta and last spark summit so last April but Delta has been an internal data bricks project for the last almost three years and so all of our largest customers are using it in particular I'm gonna walk through the Comcast's use case a little bit because they have a particularly large amount of data and discovered some interesting things but this is just sort of to make it clear like this is not you know an unknown technology or an unproven technology this is really in production receiving exabytes of data per month and it scales very very well so Comcast in particular Comcast is using spark to process all of the data they get from their cable set-top boxes so all of that data flows to kafka servers and then is eventually dropped into a data lake what Comcast's the problem they were having is when there's like a big fighting event or the Super Bowl this weekend the amount of data coming from that set-top box collection is literally going to go to a thousand X the daily average and so when it does do that they actually started ddossing s3 and they were getting throttled by s3 s3 was a s3 will send you like back-off warnings and be like you're not we're not letting you write any data you need to back off and they were getting this because they were having to use such large spark clusters to process all the data and write all of it it really actually wasn't the literal size of the data that was the problem it was the number of files they were trying to write that was the problem that was causing s3 to throttle them they were trying to write something like I don't know 10 billion files per batch and s3 was not having that so with Delta we're able to implement two things so Delta has scalable metadata so it doesn't have to load all of the data the same way that a traditional park' table will instead Delta it explicitly lists out the list of files to be loaded and so when you're reading from or writing to a Delta table it's able to minimize the amount of metadata you need to process at any given time this can also help on the compute side because you can use smaller driver nodes for your spark jobs because Delta's metadata is smaller than a traditional park' table and then additionally we sort of built in a couple of other features for them to do too in the first step right smaller files to s3 larger files to s3 so they're writing less files so we sort of like increase the file size by a hundred X so they're writing 100 like x less files and then and then on the second side of that when you do write data to Delta because it has acid transaction functionality you can easily go in and compact the data that's already in the table so you can load that data back in andrey compact it into larger files so your downstream readers don't have to make as many get requests from s3 and because of the acid transactions you do you can have your downstream readers running the whole time while you're making this compaction happen so most people who are managing spark pipelines have built some kind of compaction job in the past right but actually timing it so that it's compacting your table during a moment when it's okay to rewrite data is really hard to schedule when you have a lot of pipelines coming off of it so this makes it a lot simpler because you can dump all of your raw data directly into a Delta table and then you have basically infinite reader scalability on the read side so you can have several hundred stream jobs coming off of a single Delta table and the metadata just scales up and then because it's all reading from s3 there's no bottleneck and it just scales infinitely and we do have several customers with over 500 single streams pointed at one Delta table and so it really is just depends it leverages s3's basically infinite scalability there and then what Comcast was able to do is they were able to reduce their overall job what they had done is they were splitting up the job actually like sharding it so they had ten shards of 64 instances processing 1/10 of the data each and instead they were able to reduce that to just a single job of 64 instances so Delta Lake itself is just a library it's a specification for how to interact with the data and I'll show you in a second what the actual log looks like why the spec is needed but in order to actually get started using it you only need to be using SPARC 2.4 - or newer and then you just include the package and there's a PI spark and a Java Scala package and it's designed as a first party data source so you think of it really as a data source designed for spark and specifically targeting all the problems that spark creates and all you do is you swap over from format park' to format Delta and this is some roadmap stuff that is coming that we're working on which is basically including even more database style features the idea that I won't be able to implement expectations on my data I want to say this column I believe is a timestamp and it should only be greater you know than yesterday or and then different levels of what to do with the data so whether to fail the job if it violates it alert someone or to lambda the data and quarantine it for you this stuff is in the works and will probably be ready around the time of sparks up minute in June so what actually does Delta look like so you see here the date part right this looks like a normal partition in a normal parquet or sea table what's not normally there is this other folder Delta log and so this Delta log is a sea read series of ordered commits and each commit represents a new snapshot of the table and then inside the actual data the data itself is stored as a normal part K table the other thing is it's also very easy if you decide you don't like Delta and it's a terrible idea you can very easily switch back to just being a normal part K table all you have to do is delete the Delta log folder it's a parquet table again so there's really very little overhead and no risk to trying out Delta inside of those JSON files there are three operations that can happen you can change the metadata about a file you can add a file or you can remove a file and that's it so these invariants give it the ability to increment snapshots and alter the table State over time but it doesn't violate any of the object storage principles so we're never actually overriding any files we're only appending and then we're tracking them all the time and so this lets us circumvent a lot of the problems that object storage has we don't ever have to directories so we're never seen around waiting for s3 to tell us what's inside a folder because we've already tracked every single file that's in the table and we're adding and removing things at the file level so each commit that happens every time I write something to the table a new JSON file will appear and that JSON file will contain something like this add 1 dot park' and add 2 dot Parque or if I deleted some data it will say remove and so each writer that writes to the table puts one of these new JSON files and then what do we do when two writers are writing at the same time Delta uses what's called the optimistic concurrency optimistic concurrency assumes most transactions should be ok to happen at the same time right if you just have two separate append jobs those are always okay to allow to happen at the same time what this means is that we check whether or not there conflicts after the write occurs so a pessimistic concurrency and a database write means that we never allow two transactions to happen at the same time optimistic means we always try to let them happen at the same time and so what they will do is so like in this instance user one started at version zero and it saw that the table state was version zero user two started at version 1 and row version two but by the time user ones job finishes it's now the tables version has now incremented to two and so the writer says ok we're at a new state from when I actually started my write and then it will check is it ok for me to still commit the right I was writing and if it if it does determine that these don't conflict that's what it does it allows it to commit the data so it basically says has any of the files that I read originally have they been modified if not I'm okay to still write my data and most of the time if a job does fail it's simply safe to retry it so then how do we actually scale the metadata right I'm sure many of you are looking at this saying they're gonna be 50 million JSON files in here that's not gonna scale at all okay every 10 JSON files we check point the state to a park a file or multiple parque files if you have a big table what this means is that we're able to scale the metadata's the metadata information using spark itself so instead of storing this metadata in a database that has a limit on the number of threads it can use or somewhere else we're storing it in a format that spark can read natively and operate on with all of its natural parallelism and so now spark itself is managing the metadata of the table as a spark job and so this allows us this is what allows Delta to scale to hundreds of millions or billions of files under the table and so if you actually have that much data these park' checkpoints will end up being you know 10 gigabytes or 50 gigabytes worth of data but you're trading that off for not having to scan an entire s3 bucket looking for your data right so you can kind of already imagine Delta adds a little bit of overhead to a normal Park a table but trades off massive scalability in exchange for that overhead so if you only have one gigabyte of data you can just still use a regular part a park a table but if you have a thousand if you have you know a hundred terabytes of data you're gonna want store it in something like Delta that's doing these things so that you can scale them at a date properly and then this actually just did just come out so version 0.5 just came out improved a few things like concurrency support over time you need this function called vacuum in order to delete old files so right we're not going Delta is never overwriting any files so over time there are these tombstone files that need to get cleaned up that's what vacuum does update syntax and then one nice thing is you can convert a Delta table a Park a table in place to a Delta table so you can simply run the convert to Delta command and it traverses your park a table and builds Delta's metadata and then it's a Delta table and we also have described history this lets you show every commit that was ever made to the table and it can also record other information such as like who that user was you know whether what type of committed was was an independent was an update and both Scala and PI spark api's this project works out the box on a sure data Lake and s3 it requires like certain expectations out of the storage system we'd love for like anybody interested to be testing Google Cloud Storage or it also works on HDFS I should mention that but that's sort of like it works on HDFS accidentally that was a lot easier than s3 and this is the website as well so if you guys want real quick I can do a little demo of how the actual code works and what it looks like there are a lot concepts here but I think Delta makes a lot more sense when you completely see what's actually happening under the hood and real quick the the websites Delta IO that was available for some reason and it has this nice website and then really all the activity in the project is happening on the github where the developers are very active they've been laying out milestones for each version and there's a bunch of like open issues for things people want so we really would love to get more people contributing to this project we're hoping moving it to the Linux Foundation will unblock a lot of people contributing to it but there's a ton of work to be done right there's 40 years of database research waiting to be implemented for big data systems and there's so many features that anybody can be doing so in order to like demonstrate the way Delta actually works so I want to do is try to show how it's both a streaming data source and a batch data source what that actually means for how you can alter your workflow a little bit so what I'm gonna do is create some fake data in a stream spark has this nice fake data source called great and this will just produce data at the rose per second you give it so it's going to produce ten thousand rows per second and use ten cores and then what I'm going to do is I'm going to turn this fake data into sales data so I'm going to basically take some random numbers I'm going to create a product SKU a quantity value and a price oh yeah sorry yeah that's a great point thank you it's that big enough okay cool so create some fake sales data and then real quick just to look at it and see what it actually looks like we'll start the M start up the stream and so Delta itself right it's just a table same as any other like spark table really so you can run all of your normal catalog operations on it you have described extended other whatever the five different variations of describe bar and spark and under the hood it's just stored on an s3 bucket so this is just this is a s3 bucket right here and then inside that s3 bucket this is where I'm storing it really the only change you'll see is this provider switches from parkade to Delta now if I look at what the actual data looks like it looks like some really simple sales data it has a product SKU quantity and a price and so I'm gonna start writing this data to Delta my initial Delta table so if you're not familiar with structured streaming like the API for between regular spark and structured streaming right is just you switch right to right stream and now you're writing the stream of data instead here I'm going to partition the output data by date I'm going to append it to the Delta table in production you're always going to set a check point location for your stream and then give it the destination path and now it's going to start it I'm gonna go cancel cancel this other one and so now I'm syncing data into this Delta table well I can do is I can go in and I can do something like a show partitions on it right and I've run this before so it's gonna have an old partition in there ran this nine months ago apparently and well I can also do though is I can query it as I'm writing data to it so I'm streaming data into the Delta table right now and I can also just load it and run select star or do whatever I want it doesn't matter because when I run this command it doesn't care that there's data being written to it at this moment it's loading a particular version of the Delta table and it's querying those files that correspond to that version so it doesn't care it doesn't see the new data coming in because when I started this query it locked itself into a particular version of the table and if I want I can actually look at those versions and potentially pick one out that's of interest to me and this is something that data scientists like to do a lot right if you're building a model you don't want to build betraying that model on different data every time in fact you want to go back and retrain on the exact same data you want to retest on the exact same data or some places there's legal requirements for model reproducibility so you have to retain that data you trained the model on Delta lets you do that sort of out of the box with time travel and version tracking so I have ISM every single right increments this version by one and so I've written 910 batches for this table so far and we'll say sort of give you some additional metadata about what's happening in it Delta has these ideas of isolation levels this is sort of you know will be familiar to anyone who's worked with databases for a lot isolation levels is a concept of like what can I allow to happen at the same time right now Delta has two levels of isolation there's right there serializable and then right serializable so if you have if you're mixing the pens that are okay to mix and match you do a right serializable and so if I want to I can take this version and I can select from a particular version I think that's the syntax that's not it all right this is embarrassing actually think it's not that's time stamp as and then perhaps the more interesting thing to do is move on to the downstream jobs from this right the raw data as we said is not actually valuable in a real meaningful way yet so no this is something that actually happened to me in my last company at some point we realized that we were recording the price of each object not the total price paid because we weren't multiplying by the quantity so at some point all of our revenue numbers jumped up a lot in a big way which was awesome but actually executing that was really difficult I had to go in and overwrite like two years worth of data whereas with Delta I can do is I can actually read the upstream Delta table as a stream so now as I'm appending this data into the raw table I'm actually going to be able to update my silver table in real-time as well so as soon as that data comes into the upstream table the downstream stream is going to pick it up and immediately start writing it over here and now this data you know in in real life the other thing I would probably be doing is I would be looking up some kind of business information about the SKU itself right and so I might be pulling in from a database or like a Cassandra or some kind of key value lookup store I'd be pulling in more information about the product oh okay and then most likely what I would do next is I would take this data that is actually good for doing like product analysis product sales analysis and I'd probably start writing this data out to some kind of reporting location either exposing it to a presto or an Athena or a Cosmo DB and pointing something like tableau at it and so now I might want to you know do like a revenue by day report and I can very easily do this either as a stream as well and use something like watermarking or I can do it as a batch job and then the nice thing about batch jobs or stream jobs is because they're always picking up the latest version of the Delta table you can actually build these pipelines that join together Delta tables and they'll automatically update themselves over time so you can have like a lookup table that's a delta table and broadcast it but if you ever update that lookup table Delta automatically picks up that update through its metadata and in the very next batch will load the updated version of that Delta table and so that sort of finishes it up do you guys have any questions oh yeah yeah definitely yeah oh no no no absolutely absolutely right like I think a lot of people weave intraday data as you know like inaccurate and and would prefer to get you know accuracy I always say like you real-time data you're trading off accuracy for real-time right and you're gonna do your most hundred percent accurate metrics are going to be built by offline batch workloads typically right right so we're streaming can still help you though is on cost savings so because streaming does incremental processing right with a batch workload if I rerun the report for you know yesterday's data or last week's data like a seven-day window right I only actually need to rerun it for whatever day's got updated but I'm gonna rerun it and load all seven days of data and then calculate the same things even if some of those days haven't been changed right with streaming it's going to only check the new files and so then you can build these pipelines that are loading you know like half as much data or 1/3 as much data and so where this can really help you is actually on the storage access cost you're right you get it you're getting charged for every get request that your pipelines are executing and so if you can reduce those by 10x right your some of your storage bill goes down quite a bit as well yeah exactly exactly so yeah if I'm only having I only added one new file and we need to process one new file then I can just keep all the data I already had and just update with that one file yeah so I think especially with spark spark as on this trigger once mode in streaming and what that does is it makes the stream run as a batch workload or it just picks up all the new files and processes them in one go and then shuts down so that can be really useful where you have batch pipelines that might benefit from this but definitely like there's still plenty of use cases where it's like okay like for one a report for the last hour of data I should just run a batch workload for the last hour of data yeah so I can move it along oh yeah yeah some questions about HDFS first object storage like HDFS just behaves like a real file system so it doesn't have eventual consistency problems so it like some of the metadata stuff scales law better like you don't have folder listing problems or like that kind of stuff whereas like on s3 like if you have more than 10,000 objects in a folder the listing actually becomes a huge bottleneck so there's like Delta has a ton of optimizations they're actually built in specifically for object storage and like one of them is actually this like zero padding in the commits this forces them to get ordered in a certain way and then we can use this more efficient API list from instead of having to list so there's just like all these tiny little things like from reverse engineering object storage HDFS doesn't have these problems so it just works it's very fast on HDFS right yeah definitely definitely so yeah if you like if you compare you know a hundred gigabyte Delta table with a hundred gigabyte parquet table you partition them the exact same way you probably will see slightly slower query times with Delta in the order of like a couple seconds to a couple milliseconds so what we do is I'm when you're querying the table the first time we actually have to load the table metadata and trying to look for this yes so here like this spark job is actually loading the table metadata and then but once it's loaded it the first time it caches it in memory so in subsequent queries against that table are going to be very comparable to parquet and then we're working on a feature called data skipping and the ideas on parquet already has the min Max values of every column in the parquet footer but if you take that the problem with that though is that in order to do push down there you have to go to the footer of every parquet file still so a lot of times it's actually faster to just load the data and filter it in SPARC than it is to do predicate push down in parquet and so instead if we collect those data skipping values into the metadata about the files then we can do better query planning when you query the table we can skip a bunch of files and then by doing that then you know like you might be loading 10x less data than you would with a park a table where it's not able to do that data skipping nope I think there's a question over here okay I'll get to him next yeah yeah great question graduation so the the license is still Apache the reasons for not giving it to Apache just have a lot to do with how the foundation governs projects I don't know if you've been following with spark but spark three has been trying to come out for the last seven months now and like a lot of the problems is that the way Apache projects are governed is that major version is the only time to get a feature in and if it's not in by 3.0 it can't be introduced at all in three and so there's all these like weird problems where like everybody all the commanders are trying to get a ton of code into spark 3 because otherwise they'll have to wait till spark 4 so at this point it's been over a year since spark 2.4 came out whereas before spark would do two to three or four releases per year so like I mean some of it was just project maturity and as it gets bigger and bigger release is slowed down but at the same time a lot of it has to do with the actual project governance rules and so that's what we really were looking at especially you know Linux Foundation also owns cloud native compute foundation and so that's where kubernetes is that's where a bunch of other projects are so we just kind of saw it as like a different foundation with slightly different governance rules where we could iterate a lot faster on this project but it still sort of become a foundation that people are putting these cloud projects in and then the other thing is um it is a data source right so just the same as like park' is not a part of spark Park A's its own project so that's why it's a separate project as well deuteronomistic guarantee you guarantee that every syllable that slow transaction competing with a lot of short connections and they're gonna come right absolutely absolutely so yeah so I mean that can definitely become a problem you know like when you're trying to do updates against the table you know like two sets of updates will conflict a lot of times so one thing we do have is we have partition level isolation so if you're able to isolate the updates at a partition boundary then you can avoid that but like you absolutely could end up in a cycle of the only thing is that we don't retry that job right away so instead we just fail it and we try to detect as early as possible but you definitely could do it right delete a hundred people and at the same time I'm updating you know 10,000 people Oh so I mean app ends never conflict right right right so I mean it does depend on like the semantics of correctness here right like if I submit a deletion request for a person and then while that deletion request is happening you add a new row for that person right it's I don't remember which side we err on I do think we err on the optimistic side so we'll allow both transactions to go through but it definitely does sort of like create a little bit of so just be clear I'm not lawyer you a lawyer yeah I've read a lot about gbr too but just be clear neither of us has lawyers right like so like I did read like so what you have to do with gdpr at least from I've read like you have to actively delete from current systems and then you also have to promise to delete from backups as they become available for deletion so this is where like things start to get a little bit weird right like your s3 bucket could have retention policies so it doesn't even matter if you delete it from Delta because maybe your s3 bucket is backing it up as well we're definitely not claiming Delta to be a weight no I was right so you cannot undo a delete right exactly so the way Delta works is um so I say I want delete one bro what we'll do is we'll go in and we'll try to identify the single Park a file containing that row and then we actually write out a new Park a file without that row then separately you run the vacuum command the vacuum command deletes all tombstone files and you can't time travel past the vacuum so then that way you ensure that actually is fundamentally deleted but this is where like I actually want clarify with my with our lawyer about whether or not this is true like the way the backups part is phrased in GDP our law it would appear that we're deleting it from the current snapshot of the Delta table and it will be cleaned up by the vacuum once the Delta tables retention period is hit so I'm pretty sure in some way you can definitely work out a compliant pipeline using that so embrace with right right and so that's like another API that's in the works is also the idea of locking right which is most database is will also provide you a way to lock a table so that a particular transaction has to go through I think like this is where you know like compared to a database system it is missing the vast majority of things Postgres provides right but all of these things are don't exist over object storage natively so that's where Delta is starting to try to build these things up you get an essentially about yeah yeah definitely so especially like more and more as the clouds have nvm Hardware available that like becomes you know more and more of a solution like in particular like we've seen huge benefit just from using nvme instances for shuffle files because of the speed you get back from it it's a good staging location for putting data to object storage as well I think like that's one of the biggest challenges of these distributed systems raised you don't have shared memory that a database normally has and can can use to do these things I we are like working on features related to that but that's not something that you know we've sort of like addressed immediately No ooh that also becomes served more of like systems architecture some of this too is just sort of like given that we are writing from spark what can we do you know and how could we build a storage specification that spark would be optimized for it's sort of like I try phrases like um you know these benefits of separating storage and compute over a traditional database but you're it's actually still good for your compute to be able to expect certain things out of the storage [Music] right right yeah so so it relies on spark for the execution engine and then the the reason why I call it like really what they I would say it is is is this Delta is a transaction log protocol is a protocol for interacting with the data and reading the Delta log to then read the park' data so it really is just telling spark how to interact with this data source kind of similarly to how existing data sources already work right like partitioning in a park a table or RC table is a storage spec and spark knows how to read that storage spec Yeah right now the only way to write to Delta is from spark we are trying to engage with the flink community to build out a flink writer yeah and but for the most part like a lot of this stuff is driven by like customer requests we don't actually have any customers asking for other systems to write to this level of storage most people just want to be able to expose it to presto and hive for the most part and so the hive community worked with us to build the hive connector so now if you can create a hive table that natively updates with each snapshot of the Delta table and then soon should be the presto reader as well and then we have ways to read already and presto but you get like a static view of the table thank you all [Applause] [Music]