How to run frequent ETLs with Dagster

[ssillaots-boku]

Hoping to get some ideas if there's a way to implement frequent ETLs with Dagster's AutoMaterializePolicy/AutomationCondition.

What I strive to achieve is really simple. An ETL asset that's partitioned every 5 minutes. That's 288 partitions per day and a bit over 100000 partitions for a year. Might be okey if there's only 1 ETL asset but if you have bunch of them then using partitioned asset is not scalable.

I've seen some example where the ETL asset is unpartitioned and runs every 5 minutes. Totally doable but

• how would one set up the "connection" to downstream assets so that downstream asset materializes automatically
• how would you rerun a partition from the past
• overview of this kind of an asset is quite poor, e.g which partitions are done, which have failed etc

[sqkkyzx]

It's quite surprising to require such detailed partitioning.
For the automatic materialization of downstream assets, you can use declarative automation for downstream assets with AutomationCondition.eager(). You can refer to this documentation, which allows passing automatic materializations on the latest partitions. If you also want downstream assets to automatically materialize on past partitions, this document will be helpful to you.

I don't have such detailed partitioning. I record the partition key and materialization time in a database and calculate the freshness of a partition asset by comparing the time interval between the last materialization time and the current time, with some weight calculations. Then, I use a sensor to query SQL and re-materialize old partitions that do not meet the freshness requirements.

[ssillaots-boku]

Up!

Hoping that there are some ideas floating around.

There is an option to "half-solve" it with Asset Observations. For example an Hourly partitioned asset would run with a schedule every 5 minutes. Every 5 minutes it would ETL some data and it would mark the 5 minute chunk as completed (with Asset Observation). The partition itself would materialize when 5 minutes chunks for the whole hour are done. There's more to it code wise but that's the bigger picture.

So it's doable. But as I said before this is only half the solution.

If there is a downstream ETL asset that needs to be run in every 5 minutes as well then this approach won't work because how would the downstream asset know when upstream asset's 5 minute chunk is done. Yes, there's probably a way to build a rocket ship (sensor) which triggers the downstream but it's not scalable and I'm sure there are other corner cases which will make it extremely complex (e.g rerunning).

[v1gnesh]

Although I like this (direction), IMHO, I think a good way to give Dagster this capability is to integrate with something like bytewax.
Tagging @awmatheson @whoahbot @lfunderburk from that team for awareness... and also @schrockn pushing my luck 😓 .

It's probably a longer term project and it involves chosing which ETL provider to integrate with.
To me, bytewax is a first option as it's python (and offloads heavy work to Rust internally).

Managing ETL with Dagster, as you've found, is not fun, as it isn't mean for that, but I'm with you on the expectation that if Dagster wants to be a single pane of glass for data observability, it has to also become SPOG for "observability pipelines".
Obs is just a sub-set (or one use case) of data pipelines in general.