Scale By The Bay 2021 : John O'Dwyer, The Incremental ETL Architecture

Scale By The Bay 2021 : John O'Dwyer, The Incremental ETL Architecture

Recording: Scale By The Bay 2021 : John O'Dwyer, The Incremental ETL Architecture

hi thanks for coming today um as soon as mentioned i am i'm a developer advocate uh for databricks and uh today's talk is is uh the incremental etl architecture we will pretty much focus on open source technologies around apache spark and delta lake uh both of which are championed and pioneered by uh by databricks uh just a little bit about myself uh i'm john underwire um developer advocate before uh joining the databricks team as a developer advocate i was a solutions architect here for four years and before that i was a data scientist uh and a developer early adopter of both apache spark and also the the platform of databricks outside of work i i live in denver colorado i try to get up to the mountains and enjoy uh active at outside activities in the mountains with my family [Music] so um excuse me one sec very sorry about that all right uh the incremental uh etl architecture um the it actually really starts with uh ingest and um and essentially what what we're trying to uh do is only move data between source and destination in multiple sources and destinations but only moving the data that is either new or changed this is an idea that that really isn't very new it actually is something that data warehousing and uh and etl can do uh but the ability to do it at scale in big data environments is actually something new and really is uh really is uh spearheaded by uh innovations uh that delta lake and also uh spark is a um as a whole um help with why what are the advantages uh and these advantages are directly related to big data big data storage is inexpensive it also it separates storage from compute what that allows you to do is instead of actually holding information in in memory for for example in certain cases or anything that's like hot swappable um it it allows you to um to store the data completely separate from from compute so that uh it's extremely cheaper to actually keep that in in storage and not worry about the uh the computation that is involved with accessing that data it's also you can also retain all history which comes in very handy in a lot of cases for example we have a fortune 5 company that that uses our technology and delta and structured streaming to essentially keep years worth of logs uh for a threat detection and uh the reason why they use uh this particular uh type of architecture is so that they can keep all history instead of only keeping say 90 or 30 days worth of worth of information so they can do uh uh threat threat analysis over the course of years it's also efficient uh the reason is is quite uh quite simply you're only you're only processing new or updated data as opposed to reprocessing information that that ancillary needs to be changed in object store let's say you have a an object that actually holds thousands of records and you only need to change one in times past you actually needed to essentially this is simplifying it a little bit but i you would need to blow away that whole file or maybe even a whole partition and uh change that whole set and and then update the values and the whole time that you're doing that you couldn't have other people hit the information at the same time that was very efficient um the uh the other thing that's uh great as an advantage is that you can actually have these as multiple data sets for example this bronze silver and gold architecture it it looks linear but it doesn't necessarily have to be and the end results can be different uh there's uh generally speaking these gold tables are either for uh aggregated data sets that are uh used for business intelligence and analytics or featured feature sets that are used for machine learning whether it be training models or inference after the training is done and this this isn't necessarily usually linear uh you would have almost a dag of uh bronze silver and gold tables over the course of of everything um other things that make this uh advantageous uh these uh these tables can be uh are atomic and always available as i mentioned before um they it is uh it used to be very inefficient if it if a value needed to change and only one value and a table needed to be changed sometimes you'd have to blow away that whole set um uh the the date the the transfer of the data uh is also stateful and you can um and we will get into how how this is stateful but essentially we keep track uh we being uh uh um structured streaming keeps track of where where uh the state is at any given time and uh and um uh the um the that includes stopping for jobs or errors or the like um and in the case of uh latency uh you can actually change the cadence of of these uh of of these runs uh structured streaming apache spark structured streaming specifically is made to be continuous so 24 7 and extremely low latency so that essentially information can be picked up right away and move through the system the uh but at the same time and we will get into how to do this they can also those same jobs can be changed from continuously streaming to schedule uh using something called trigger once where instead of just running continuously you can you can let the uh the system identify when it's done with the information or processing the data that's associated and it will cut itself off and uh then you can then you can actually run these on a schedule say daily monthly hourly and the really nice part of that is you can actually drop the cadence over time so that it can be continual if you like since since storage and compute is also um uh separated uh you can essentially keep historical data sets uh there is also uh the ability to version uh in delta delta lake tables are versioned and you can always go back to historical data sets and reproduce uh reproduce things like specific ml models directly from the old data sets as they were historically so why is this hard now well the reason is that and it's not necessarily just now these technologies like data lake uh delta data delta lake are actually have been around for a little while but preceding them the only way to really do incremental etl was through warehousing data warehousing where you can actually on a row-by-row basis pick and choose what needs to be changed but that that is expensive a lot of data warehouses also do not separate compute and storage um updating data in big data was not necessarily possible and state is hard to hard to uh to keep track of uh there's a there's a term called item potency and essentially that's a big word for uh pick up where you left off so uh essentially this state will um will actually do exactly that where if if something errors out if something has finished and needs to restart instead of keeping track of where you left off or re-recomputing old data that may or may not have actually gotten in and then dealing with whether that information is duplicative or uh or if it needs to be reevaluated essentially structure streaming takes care of that by like just out of the box and then also you know you're only you're only uh computing what rows have changed and that's uh instead of the inefficient manner of doing everything and under trying to understand what has been uh what has changed and what hasn't so um before going on i just wanted to see if uh sundeep if there were any questions uh i can i can take those if not we can actually save a little time at the end for um for a little demo um but if there are any questions that you think that we should answer um let me know yeah no questions you're free to continue sir okay great thank you so um let's step through the different pieces of the of incremental etl and uh in starting from left to right well actually starting on the left uh is uh is ingest um there and this is actually the piece that uh generally is outside of apache spark structured streaming and also delta lake and it's around uh streaming ingest or uh cubase uh cube-based ingest um these are continuous and or or scheduled uh ingest the difference is again based off of uh based off of um if if you want to identify when something is finished and for the for the stream to shut itself down or not uh and we'll talk about the functionality of what that looks like in a bit popular modes of ingest are kafka that would probably be the main open source technology that people use for this others are kinesis on aws event hubs on azure and then cubase systems uh would be sqs which is on aws databricks has a technology called autoloader that's great for this in the case of autoloader you can actually drop values directly in this into object storage on the major cloud providers and it will identify which in identify in order uh which data needs to be evaluated first and the the use of autoloader actually removes the need for you know an extra layer of technology and functionality uh for streaming so um as i mentioned there's there's two main components to two incremental etl those are apache structured streaming which we'll discuss here and the attributes of it that that lends itself to uh incremental etl the other being delta lake and we will discuss the the features that that we use in incremental etl in a second as well uh i had mentioned the term item potency uh that is directly related to uh the uh checkpoints in state and how it's saved um in this green part here it's actually the bottom the second line of the green is uh is where you specify a checkpoint location and that will automatically actually save where you've stopped writing at uh i'll i'll explain this as well what we're doing in these two lines uh you actually have a read stream and a write stream so the read stream actually reads from a place in this case it's reading from a format called cloud files which is uh the autoloader format into a streaming data frame and then uh the right stream writes it out in the case here into a delta table so this would actually be writing from ingest to a bronze table and using uh this checkpoint location essentially as it's writing the metadata that's associated to the state of where it is at any given time uh what that allows it to do is when when the stream identifies that it's finished uh uh reading data it can shut itself down and it will that checkpoint location will understand where to pick up when the next run needs to happen this is actually great for uh for errors as well as we all know there are foreseen and unforeseen exceptions and errors that happen in uh in data etl and um and you know these types of architectures is uh this is uh no exception and uh and this helps in that case the other part of structured streaming that uh that is very great in uh in incremental etl is is uh the ability to go between continuous and scheduled uh in this case ingest but we will talk about continuous and scheduled uh traversing of data between the table between delta tables as well and how you do that is uh the use of trigger ones trigger once is uh the first uh the first green line here uh you don't have to put this in if you don't this this uh this combination of a read stream and write stream will be continuous and it will just continue to go and it won't identify uh to shut itself down totally fine uh for some uh for applications that need to run 24 7 many and most do but a lot of the times what happens is um a lot of people want to be able to do this in a schedule because there's two reasons you might want to either you might not have enough data to continually run or you or you may want to not be burdened with a continuously on 24 7 stream where you could actually you can actually identify and administer it a lot easier if it shuts itself off one of the main reasons why delta lake is a great is a great big data table for uh for incremental etl is literally the ability to update delete rows in a in a in a table this really wasn't possible before on big data object stores delta lake makes it possible there's a lot of uh there's a lot of uh back and forth and reasons how this happens probably will you've probably seen other presentations on the inner workings of delta lake and uh there definitely are um but for what for this conversation really just going to keep it high level one thing that i did want to point out here is uh specifically the merge into statement because uh update and delete statement sets those are pretty standard and you've probably seen those before emergency statement may not be quite as uh familiar um this is uh this these merge statements or sometimes they're called upsearch statements they make it possible to in a very concise way uh delete update or insert all at the same time given one table this comes in handy when uh when doing when doing incremental etl using structured streaming i will uh show you an example of this actually later if we still have time so um the other thing that uh that delta lake the the features that uh allow delta lake to be great for incremental etl of big data at massive scale is uh is the movement itself um mostly that it can be a streaming source and sync so in the case of uh of moving data from a bronze delta table to a silver delta table uh both the uh in that case both both the source of that stream and also the sink which is the silver table are both delta tables and there is no need for uh other types of streaming technologies to be in the to be in between them it actually makes it a lot more seamless and easy to deal with that this is also built into the acid transaction nature of it of these tables that being that as you're as you're uh as you have outside source or outside uh consumers of these tables uh they can actually read these tables in real time and uh get the most up-to-date uh versions of those tables based off of the asset transactions and those those transactions also make sure that none of the data is corrupted at any given time uh the other thing that actually uh really helps here is that the delta tables are uh do have a schema that is enforced and also can evolve at the same time what i mean by that is let's say you have a bronze table that has five columns and your ingest is given a a fifth column um that is outside of the the initial schema uh you can determine if you want to enforce that that schema and what you want to do with that new column you can the defaults actually just to to append it if you are allowing for schema evolution there's other various cases as well you can even you can even fail if you see data that you hadn't seen before then reevaluate if you want to actually include that data it makes it very flexible this is this flexibility is really key to the business outcomes reason being is that you can um you can you can allow you can allow new data to come in and evaluate what that new data is without corrupting uh the column uh the column sets and the schema um in a way that wasn't really possible before um parquet at his core it's a great technology that allows you to actually evolve but these schemas in the case of parque they can evolve in uh in ways that will corrupt the data because they are atomic to a field instead of holistically looking at uh at the whole table so um that's uh that's the end of my uh presentation uh we do have a couple minutes um i'm uh i will be available uh in five minutes in the uh q a section uh but i did want to uh since we did have a minute or two i did want to go into the incremental into an example of incremental etl specifically i wanted to actually show you how how data can be added and updated at the same time in a streaming manner and how to identify the changes based off of that in this this this is a [Music] clearly you would be dealing with much larger data sets in this case we are going to include a set of six rows and then another set of six rows the first set of six rows are earnings per share um earnings per share uh and uh or estimated earnings per share by analyst for uh march 1st 2021. i'm going to skip down to uh the secondary set that i'm going to include and it's going to include all of the estimated uh earnings per share by analysts for these stocks for april but as you can see there's one row that's actually updated uh a value for for march so we have to figure out how we can accept that information into the bronze table and also then progress it into a uh into a silver table that is uh that's clean and then can therefore actually go down downstream and be updated even further if needed um so which is a very powerful idea um think about it in this way like if um in in one things need to be updated that's that's one problem but two even past that models change feature sets of models in the case of machine learning those also need to be changed and updated over the course of time if you can update these values instead of blowing away a whole uh set and also do that in an incremental fashion so that it actually daisy chains all the way through your uh through your etl um that is going to be a lot more seamless a lot more efficient and also really really open up uh architectures that weren't possible before for example having multiple different types of gold tables at any given time and also the ability to update those gold tables in real time so as we can see we have one extra row this is uh the essentially uh what we saw before with the read stream from uh auto loader uh and then to a right stream we're actually doing this in trigger once and as we can see in here uh in the bronze set we actually have two two different values for earnings per share for analyst 2 on stock a so this uh 2.4 is a new value and this is the old value that we are actually replacing so what we can do and i know i got two minutes i'm going to show this real quick and then we can actually jump out what we can do is uh i'll come back up to the upsert uh to delta but as you can see here we have uh we have a read stream uh we drop the rescue data column from the bronze table and then we're writing that silver table into uh into the silver table but we're actually going to go through a for each batch this 4-h batch function is where all the magic happens and this is iteratively identifying uh in that new table based off of that micro batch what values in that micro batch need to be merged into that table where whether it is in this case an update or an insert we don't have a delete case here in this use case but um but in this case we're we're in real time and streaming updating a uh updating a delta lake table using a merge statement and uh that's uh that's what i wanted to show you today thank you very much uh if uh if you have any questions i will be in the q a section of spatial chat and happy to discuss any of this or anything else that you really want to talk about to tell you the truth thank you