Orchestrating Dynamic Data Flows, AgileData Engineering Pattern #1 - Episode #67

Welcome to the Agile Data Podcast, where we talk about the merging of agile product and data patterns in a simply magical way. Welcome to the Agile Data Podcast. I'm Shane Gibson.

And I'm Nigel Vining.

Hey, Nigel. Today we're trying a new thing. We're going to start a series where we effectively describe one pattern in each episode that we use. And I'm thinking of it as pages in our pattern library. The first one we want to talk about is the way that we orchestrate the data flows in our way of working. So why don't you start off by giving me an overview of the patterns that we use to engineer that particular task?

So we do a little bit different. After many years background using the likes of Composer and Airflow to deploy DAGs, and create run manifests, we thought we'd be a little bit different and use a context-driven generate the manifest on the fly approach. So effectively what we do is we create our data rules for transforming loading data. We put them into our context database. And then what happens is the first piece of data that turns up effectively triggers a refresh of that data table. And then this is where it changes from the usual. So as soon as that table's loaded, we then pause and we do another lookup of the config. And we basically say, what are all the downstream objects that hang off this table now that it's loaded? And from that, we build a manifest on the fly. So we effectively say, great, we need to load— this table depends on this one, so we will load it first. So we build that manifest and then we seed off all those jobs. We basically set them off all at the same time. We attach a tag to each of them and then we wait for them to finish. They all run in parallel. And as each one finishes, it checks that manifest and goes, am I the last?

Nope.

Am I the last? Nope. Until we effectively get the driving table, which would start the next flow. So once that table comes back and says, yep, loaded, we do another config lookup and we start again. We create another manifest and we keep repeating this until we've effectively got nothing left in the manifest. The beauty of this approach is in our early days, we did a lot of rebuilds and changing context. Shane would introduce new context while jobs were already running. He'd delete tables or he'd recreate them with a different name. So the whole time, what was happening is we were getting lots of failures until we went to this pattern of context and manifest on demand, where effectively the whole time, it was always checking and saying, oh, new piece of configs turned up, got another table to load. And it effectively slotted into the mix, and it was all quite seamless and painless for me. So that's our pattern in a nutshell.

And the reason I like it is because if we think about each of our blobs of transformation code has been a rule, that's the term we use. I can go and add a new rule. So I can say there's a table sitting in history, think raw bronze, I need to do something nasty to the data to bring it through into our designed layer. And I can go and add a rule. I can add a small piece of logic that takes that data and moves it into our design layer. And as soon as I do that, the next time there's a scheduled execution or an execution that's triggered for a load, the manifest, the DAG, directed graph for that flow, will dynamically rebuild itself. And so it just self-heals. And then as you said, when I break my own rules for naming conventions, I can go and rename things. And again, that the data flows self-heal themselves. So effectively I can do the work I wanna do, and I don't need to care about what the directed graph for those data flows are. I don't need to care about the dependencies because the context engine, this pattern we use for orchestration of these data flows is doing it for me, right? I just don't care. And importantly, it's doing it for you because you don't care either, right? There's effectively, we can make any change we want and the machine just takes care of it. So I think for me, that was the biggest value. If I think about the alternatives, so one of the core patterns we use, which you can talk about in another podcast, is this idea of context-driven. So everything we do, every piece of logic is stored in a repository and everything that needs to execute is hydrated or generated dynamically from that repository. So what that means is when we make a change, we can dynamically regenerate those data flows versus the old way of doing it, which is we'd hold that context in one place and we'd have a physical instantiation of it as a set of DAGs. And now I have to sync it. Yeah, I have to like, I wanna change the rules. Okay, now I need to change the DAG. And those things used to always get out of sync. We were kind of like, That's not the way it's meant to work. Oh, shit, there's this piece of code stored as a stored process somewhere as part of that DAG. Well, the DAGs were a stored process, and that syncing problem used to be a real problem in the old patterns.

Yeah, and even today, I occasionally struggle with consistent DAG deployment if there's multiple people, engineers, developers making changes in that space. You can effectively get a little bit of drift of what should be running and what's associated with other things because there's a little bit of change going on as commits are happening. Whereas this way, we effectively leave it to runtime. And it's at runtime as an object loads, we look up what's next. So it's always the freshest piece of lineage or manifest that you can have for any object. It's always up to date. Because it's always doing a query back to our context layer to say, hey, what's next? And then it starts next. And that's cool until that's finished. So I like it for that because it takes care of itself.

And also all of the maps that we generate, so our data catalog, our lineage graphs, everything else is driven off that context versus the previous generation where we would've treated all the code and all the schemas and all that structure and all those schedules as exhaust. We would have sucked it into the catalog to try and render it. We flipped that model, which is you define the things in context, and then everything else is generated off it. It's always up to date because I cannot do a piece of work without the context being entered. I cannot create a transformation code. I cannot create the idea of a table. I cannot create the idea of a schedule. I cannot orchestrate dependencies across tables without first creating that piece of content. So again, that forces us to make sure that anything I'm looking at is the latest version because there is no choice.

Yeah, yep. And I guess the natural question that others would ask in the profession, what about when a change is made that breaks something? And this is an interesting one because effectively when any changes are made, we validate them and we don't allow updates to the context layer unless it's a valid update. So every change that's made won't update the latest production context until it's valid. So we effectively stop anything getting into the context layer that's going to cause a problem. It stays at a draft effectively until it's validated. So when the manifest is generating itself, it's always looking for the active record, and it's fine for this to be draft floating around. It won't touch it. Until it's valid, and then it flips over to replace the current production record.

And then, there's a bunch of attributes for that context. So when I define a table or a piece of code, I can define that it's what we call auto-sync, that it will execute that manifest, that data flow whenever a row turns up in one of its dependencies. Salesforce sends us some records, it hits that history tile or table, so Raw Bronze, then if it's set to auto sync, it will then trigger the hydration of that manifest and run it. But if I flip it and say that for that piece of code or that table, it's manual sync, custom sync, and I say it's going to only run at 7 AM on a Monday, then that's when it triggers the hydration of that manifest and running that code. So there's a bunch of attributes that are actually really important when you start moving all your data flow stuff back to a context layer.

Yeah, absolutely, and that's a good point you touch on, effectively allowing that orchestration to start based on a record turning up or a file turning up or a schedule that's been attached to it that says run it daily at 7 AM, which is quite a common pattern for pipelines. But 99% of our pipelines tend to start when something arrives, file some data, which effectively triggers the pipeline to start and pull that data all the way through the layers.

And then the other pattern that we use a lot in there is this idea of pub/sub. As you said, it's a fire and forget. It figures out all the steps, it fires them off into effectively a queue, it waits for the queue to run. So again, as you always do with patterns, you're using other patterns to make this pattern really efficient and effective.

Yeah, so the driver, as I said at the start of I'm done, are you done, are we ready to run something? Those are all pub/sub messages. So we have a standard pub/sub message which basically says, I'm done, and that's attached to everything in the execution layer. So anytime a table's refreshed, we get a, I'm done message off the end of that. And then we basically look for those messages as we walk through our layers. So that's our, basically, I guess our wrapper pattern, which holds it all together. Has something happened? Yep, cool, trigger something else. And we just use messages.

And then what's our average runtime? It's always dependent on, we use BigQuery under the covers and Google Pub/Sub, and it depends on how much data on that. From what you are seeing with all the different customers we've got and the volumes of data, what's an actual execution for a standard data flow, way to go from arriving data in history all the way to data being consumable in that final layer?

It's generally quite fast. We have very little latency because there's no overhead. And like an Airflow example, there's no overhead of starting up, I guess, execution pods, Kubernetes pods to run a task, to shut down, to start up and run another task. Because we're effectively just using a message, we can tell BigQuery to run something very quickly and in parallel as much as possible. So our execution times are generally very fast because there's zero overhead. We're basically just telling BigQuery, here's a list of things I want you to do and letting it do it.

And our costs are low because we're not running a container, Kubernetes and EC2, anything like that permanently to run the orchestration or the scheduling engine because we're using Pub/Sub and Cloud Functions, but moving to Cloud Run, it's doing all that magic for us. So it turns itself on, the code's deployed, it's run, it's then destroyed, and we only get charged for that period of time that it's actually running rather than have a 24/7 service sitting there to just schedule and orchestrate our code.

Yeah, that's exactly right. So we have no concept of an orchestration server running in the background. We are effectively creating the BigQuery jobs on the fly. Letting it run them and waiting for that I finished message to come at the end of it. So there is never a service that is physically waiting for a job to run and costing us money. Basically, BigQuery is running jobs, and when it finishes, we get notified and we automatically start the next job. So it makes it very economic to run, and we can run hundreds of jobs, thousands of messages, but it basically costs us nothing because there's nothing running to orchestrate that. Those are just messages bouncing around, starting jobs, waiting for jobs to finish. Very effective.

But there is an anti-pattern for this because we only use this for what we call micro-batching. We think about anything above 15 minutes of refresh latency, this works for. Anything below that, we start hitting a problem. And it's not a problem of pattern around this orchestration of data flows, it's a problem of the dependencies of everything else we've built. So if we think that we have a layered data architecture, history, design, consume, we use tables under the COVID in BigQuery. So there's a dependency on who's writing to them. And we know we can orchestrate this stuff very quickly, but if we move to a streaming model, you know, if we wanted to pick up one row and push it all the way through this, we would actually end up slightly re-architecting the way we deploy the code to Pub/Sub. We wouldn't actually fire off 10 tasks and wait for them to daisy chain and orchestrate themselves and have dependencies. What we'd probably do is look at the entire piece of code for the end-to-end orchestration and then submit that to run as a single code stream in BigQuery, end-to-end as one task, wouldn't we? Because the pattern we've got naturally breaks the task down into nodes and links. It says, run these, and when they're done, run these. and then if we keep firing it with volumes of streaming data, we will actually end up having a bottleneck, won't we? So that's an anti-pattern. We would use the same context approach. We would still hydrate and fire, but we would slightly change the way the pattern is deployed at runtime to handle that streaming volume.

So in that case, we would let data stream constantly into the landing area, which is what we do in some projects. And then effectively, I guess our compromise around that one is then we would micro-batch at 15-minute intervals from that stream data and move it through in micro-batches currently. Otherwise, yes, we would architect to stream through the layers and we would probably run a two-speed pipeline in that case. We would micro-batch at say, at maybe 15-minute intervals, but then what we would also do is we would stream around the outside data directly to the consume layer for the required attributes that need to be consumed in real time. And then we would catch up the rest of the layers of our slower data by bringing it through the normal pipeline, and then we get the best of both worlds.

Effectively adopting a Lambda architecture, right?

Yes.

But the majority of our patterns would survive for this orchestration of data flows, but we know that we actually have to change or tailor or augment some of them when we hit that anti-pattern for the way that we're currently running it and we're currently designing it. So apart from a little bit of tweaking for streaming, is there any reason you wouldn't use this pattern in use cases where you think this dynamic hydration of the data flows and this creating a manifest and this pushing it off to fire and forget and run, is there any reason that you wouldn't use this pattern?

No, we've seen it run. Day in, day out for coming up 5 years now. The original architecture and code that was deployed back in the first year is largely exactly the same. We have not changed the core pattern, so it's run happily across multiple customer use cases for 5 years. I think that's a fairly good testament to it's a pattern that works.

I think the other thing we can do is we can, if we needed to, we can scale horizontally or scale vertically. If we ever do get to a situation where we need to reduce the latency time of the end-to-end process, we can scale up the BigQuery instances and make them execute faster. We could chunk the data flows down into more domain-orientated buckets and then figure out how to parallelize the runs of them. So again, we've still got lots of headroom of how we can iterate this pattern if we ever strike a cost or a performance or a complexity issue. So we know we're always going to be iterating it. But like you said, it's been fairly good and bulletproof. I think the other thing is you've spent quite a lot of time on what we call the blast radius. As you started building out that core pattern as code, and as I started doing the things that I would expect to do as a non-data engineer to data, I should be able to do this, and you suck your teeth going, oh my God. You've built those safety nets around that pattern. So if you talk about the fact that I can create a piece of context, but it's not actually treated as part of that production data flow, it's not executable when it runs until it meets all these checks. We're effectively bringing in the peer review process, the Git review, the PR pull request, the data quality tests, all of that's happening before any of those things allow the data flow to actually execute as a new version.

Yeah, and so the other thing that's wrapping this is in the context layer, we've got a couple of flags that are applied to the context, so If something has changed in the context layer and successfully deployed into production state, we effectively put a flag on it that says this config has changed. It's going to require changes of downstream objects to flow this change. For example, you've added a new column into a table, and now you expect that column to flow through and hydrate in all the layers downstream of it. So effectively, we tag that object and say, hey, this has changed. You're gonna need to recreate it and you're gonna need to check all the things downstream of it to make sure that they're gonna be okay. So when we come to run that manifest, we run it as usual, but effectively we hit an object and say, ooh, there's a flag on this that says it's changed. So we identify the change. We usually recreate the table and we roll the watermark back. We reload that table. That's cool. We get to the next one and we say, great, something upstream of me has changed. I need to make sure how that's going to work. So what we actually do is we do another validation of that context to make sure it's 100% going to work with that change. And at that point, if we determine, oops, this is going to have to be rebuilt as well, we do that. And we keep doing this all the way to the end of that. Manifest. But we also stop the manifest if for some reason we've actually caused something that's not going to keep working. And we stop and we raise an alert and say, hey, this context is now going to be invalidated because something 2 or 3 layers upstream has changed enough that I can't fix it for you automatically. You're going to have to look at the change and then basically say what you need to do. So that was quite a breakthrough because we used to spend a lot of time manually trying to fix pipelines that stopped for small things like column being added or removed. Whereas 99% of the time we can handle that automatically just by doing a full rebuild and refresh of the table and then revalidating the rules. And generally everything takes care of itself, 'cause we wanted it to be self-healing, 'cause there's nothing worse than pipelines that fall over just because you've added a column. Whereas something like that's easy. A column can just be added, populated, and away you go. So that was quite a breakthrough, basically putting those extra tags in the context.

Yeah. And effectively they're like stage gates. It's basically every time it's validating, am I still around the next step? And then in the beginning, like you said, we had lots of times where it stopped. I would do something and then it would say, oh, actually you're not meant to be able to do that. But as each one of those happened, we'd go in and we'd either fix the way I created that context so I could no longer create it in an invalid way, or we'd automate the rebuilding, the deploy and destroy model that it was trying to rebuild so it could rebuild itself safely. So, again, we didn't change the core orchestration pattern. We changed all the other things around it to make sure that pattern was bulletproof and ran, and we didn't get called out at 2 o'clock in the morning because a data flow had stopped, or at 7 o'clock to have a look at it and have a whole lot of catch-up work to rebuild it. So, I think those supporting patterns were as important as this core orchestration of the data flows by them being dynamic and using manifest pub/sub.

Yeah, that's exactly right. It makes it all quite robust. And as I said, generally the pipelines run very accurately without a lot of tension. Excellent.

All right. Well, I think that one's done. So orchestrating data flows in a dynamic way using manifests and pub/sub. Right. Another one next week, but for now, I hope everybody has a simply magical day. And that, Data Magicians, was another Agile Data Podcast episode. If you want to learn new patterns that you can apply to craft your own Agile Data way of working, head over to agiledata.io.