It's about time: “event_time” + “microbatch” + “sample”

[graciegoheen]

🏺Background

Does anyone really know what time it is?

Over the years, the number of time-related configs in dbt has grown.

There’s updated_at (for snapshots) and loaded_at_field (for sources) and agg_time_dimension (for semantic models). You can configure the physical layout of time-series data with partition_by on BigQuery + Databricks/Spark, cluster_by on Snowflake, or sortkey on Redshift. You can also fine-tune how dbt incrementally processes the arrival of new data by configuring incremental_strategy, which may require you to define a unique_key (which is often unique, but not always! and subtly different from unique_key for snapshots).

😵‍💫 If that makes your head hurt… you’re not alone! @dbeatty10 keeps a collection of temporal terminology here.

We think a gap remains: dbt lacks awareness of the “event time” of a given model, the field that indicates at what time a record of data occurred. This gap is one we're keen to close — because we also believe that "event time" awareness will unlock a whole slew of features in the dbt that can save you time & money.

🗺️ We’ve got big plans for big datasets

@graciegoheen acknowledges this is the wrong “bare” but refuses to change it

Imagine this: As a dbt user, I configure an event_time column for my large models that represent time-series datasets. This is different from metadata fields about when the data was loaded, transformed, or otherwise processed — it’s the semantically meaningful column that says, “At what time did this thing actually happen?” That thing could be an order, a survey response, a subway ride, anything. If it happened, it happened at a time, and you tell dbt which column represents that time.

In exchange for configuring event_time, I will unlock two new capabilities:

• “Microbatch” incremental models for batched data processing
• “Sample” mode for faster dev & CI runs

…and more! Once this configuration exists, it can be leveraged in all sorts of places, including metadata integrations and quality extensions. For example, a few months ago, we announced “Advanced CI” that’s coming to dbt Cloud. If you’re only building 3 days of data in your CI environment, then dbt should know to apply the same filter on the other table too — ensuring an “apples-to-apples” comparison, and a much faster query.

❓Why only “time”?

We believe time-series datasets are by far the most common type of very large dataset for dbt users in the wild. We thought about abstracting this into something more general-purpose (such as shard identifier or segmentation_key), but we’d rather start with time because it’s semantically meaningful — it enables dbt to reason about “last 3 days of data,” and to know that a given time-slice (e.g. 2024-08-30) is referring to the same slice of data in every model that configures event_time.

⭐ [Now] “Microbatch” incremental models - coming in dbt Core v1.9

Incremental modeling in dbt today

Incremental models are, and have always been, a performance optimization — they seek to approximate the same behavior as the table materialization, just faster and more cheaply.

You start by materializing as a table. When that query gets too slow, you can optimize it:

• Swap the materialization to incremental.
• Add a snippet of dbt code (in an is_incremental() block) that uses the already-existing table (this) as a rough bookmark, so that only new data gets processed.
• Optionally add in a lookback window for safekeeping (and for late-arriving events).
• Pick one of the strategies for smushing old and new data together (append, delete+insert, or merge).
  • And if you go for delete+insert or merge, you can sort of control which “batch” is overridden by setting a unique_key (which isn’t actually unique - it’s the “upsert/merge key”).
• If anything goes wrong, you can always full-refresh, by running the same simple query that rebuilds the whole table from scratch.

As an example:

{{
    config(
        materialized='incremental',
        incremental_strategy='delete+insert',
        unique_key='date_day'
    )
}}

select * from {{ ref('stg_events') }}

    {% if is_incremental() %}
        -- this filter will only be applied on an incremental run
        where date_day >= (select {{ dbt.dateadd("day", -3, "max(date_day)") }} from {{ this }})  
    {% endif %}

For this incremental model:

• “new” records = records with a date_day greater than the maximum date_day that has previously been loaded
• lookback window = 3 days to account for late-arriving records
• when there are new records for a given date_day, the existing data for date_day is deleted and the new data is inserted in

This pattern has gotten us pretty darn far! Every day, there are tens of millions of dbt incremental model refreshes. But there’s also a clear limitation with this approach: Some datasets are just too big to fit into one query. It’s impossible to imagine rebuilding the whole table in one fell swoop. In these cases, “incremental” isn’t a performance optimization — it’s the only way to roll.

This is something the community has known about for a long time. Way back in June 2018, Claire wrote an experimental “insert_by_period” materialization that is still used to this day. (For a long time, it only ran on Redshift; in recent months, our colleagues have also implemented it on Snowflake, BigQuery, and Databricks.) The big idea: Rather than processing all data for all time in a single query, the data should be batched up into “periods” (day, week, month) that are each processed in a separate query.

How to define those batches? How to template the appropriate filters into model SQL? As an experimental materialization, insert_by_period offered some creative answers. But what if the answer just felt like dbt? This isn’t a new idea; its time has finally come.

The future of incremental modeling in dbt

Epic: #10624

There are limitations of the current incremental model approach:

• The burden is on YOU to calculate what’s “new” - what has already been loaded, what needs to be loaded, etc.
• This can be slow if you have many batches to process (like when running in full-refresh mode), as it’s done in “one big” SQL statement. In theory, the data platform can optimize processing of distributed datasets; in practice, that query can time out, and when you retry that query you end up reprocessing the same batches.
• If you want to specify batches for your incremental model to process (”rebuild for 2024-09-01”), you need to add your own home-rolled logic, likely using vars

In view of these problems, we are planning to implement a brand-new incremental strategy to handle this type of batched incremental processing out-of-the-box. Introducing: incremental_strategy=’microbatch’.

Let’s take our same example from before, and instead use the new microbatch incremental strategy. This is the proposed spec, liable to change slightly as we implement:

-- proposed spec
{{
    config(
        materialized='incremental',
        incremental_strategy='microbatch',
        event_time='date_day',
	batch_size='day',
	lookback_period='3'
    )
}}

select * from {{ ref('stg_events') }} # this ref will be auto-filtered

Where you’ve also set an event_time for the upstream model(s) - in this case stg_events:

models:
  - name: stg_events
     config:
       event_time: my_time_field

And that’s it!

• You write your model query for exactly one day of data. You don’t need to think about is_incremental filtering or DML (upserting/merging/replacing) - we take care of that for you.

  • dbt will generate the where clause(s) for you, using the event_time and batch_size configs from the incremental model and upstream models

• During standard incremental runs, dbt will process new batches according to the configured lookback_window (with one query per batch)

• If you have configured an event_time for upstream models, we we automatically apply a filter. (If there’s an upstream model that configures event_time, but you don’t want the reference to it to be filtered, you can specify ref('upstream_model').render() to opt-out of auto-filtering)

• To load a specific batch, you can pass your event time upper & lower bounds at run time using start_event_time and end_event_time - dbt run --event-time-start "2024-09-01 00:00:00" --event-time-end "2024-09-04 00:00:00”

• Should a run of a model using this materialization be interrupted, a subsequent run will continue building the target table from the batch where it was interrupted.

Question: What does “full refresh” mean for microbatch?

If a model is big enough to be microbatched, you probably don’t want to accidentally replace every batch just by typing -f!

We are considering among three options:

• dbt run -s my_microbatch_model --full-refresh will rebuild all batches for all time — similar to classic incremental models, although each batch will have its own query, and those queries will run independently.
• Microbatch models should completely ignore the --full-refresh flag. Setting full_refresh = false (docs) should be best practice, or the built-in default.
• “Cold storage”: Microbatch models should --full-refresh up to a point. If standard incremental runs reprocess 3 days of data, then full-refresh runs should reprocess 30 days (or a period you configure) — everything beyond that is in “cold storage.” It’s still possible to restate historical data beyond that point, but you need to enter “full control mode” by passing explicit start_event_time + end_event_time.

🌀[Next] “Sample” mode for dev & CI runs

In v1.8, we introduced dbt run --empty, which enables “dry runs” in most dbt projects without any code changes. Whenever dbt finds a {{ ref(...) }} in an “empty” run, it will template select * from {referenced_model} where false limit 0 so that the query completes without processing any actual data. This is also useful for unit tests, which introspect the structure of upstream objects in the data warehouse so that test fixtures don’t need to specify every single column.

We want to support a similar pattern, for faster development and CI testing against a time-limited sample of real data (first described in #8378). Instead of empty inputs, dbt would automatically filter down large tables to the last X days of data. How would it know which tables are large, and which columns to filter on? Why, event_time, of course!

Let’s say you have a query like:

-- my_model.sql

with fct_orders as (
    select * from {{ ref('fct_orders') }}
),

fct_page_views as (
    select * from {{ ref('fct_page_views') }}
),

dim_customers as (
    select * from {{ ref('dim_customers') }}
)

select ...

During standard runs, this will compile to:

-- target/compiled/my_model.sql

with fct_orders as (
    select * from my_db.my_schema.fct_orders
),

fct_page_views as (
    select * from my_db.my_schema.fct_page_views
),

dim_customers as (
    select * from my_db.my_schema.dim_customers
)

select ...

During sample runs, it will be possible to configure a single environment or run-level variable that instructs dbt to check each upstream model and see if they define event_time. Let’s say you’re working in development, and you want to build your models using 3 days of historical data by default. The exact syntax here is TBD, but imagine it looks like something like dbt run --sample 3, producing a query like:

-- target/compiled/my_model.sql

with fct_orders as (
    -- the event_time column in fct_orders is 'order_at'
    select * from my_db.my_schema.fct_orders
    where order_at > dateadd(-3, day, current_date)
),

fct_page_views as (
    -- the event_time column in fct_page_views is 'session_start'
    select * from my_db.my_schema.fct_page_views
    where session_start > dateadd(-3, day, current_date)
),

dim_customers as (
    -- there is no event_time column configured for dim_customers
    select * from my_db.my_schema.dimension_table
    -- so there is no filter applied
)

select ...

You will be able to set the sample flag in CLI commands (to modify specific invocations), as an environment variable (to modify dev/CI as opposed to prod), and in your project file (to set a default behavior for everyone).

A “sample” here means a consistent time-based “slice” across all models. Meaning, for a given run or environment, dbt is filtering down to the same number of days everywhere. We’re sure there are reasons to want more-configurable samples — a tenant/country ID instead of time-based, or different slices for different input models — but we also think this simplification can be powerful for user intuition sooner, and allow us to do nicer things later on. How far can we get with this simplifying assumption?

What is the overlap between “microbatch” and “sample”? It’s simple: Both depend on you configuring event_time so that dbt knows how to filter down your inputs. Once done, they can form a two-part harmony: If you dbt run -s my_microbatch_model --sample 3 (exact syntax TBD!!), then dbt will build that model for 3 days of data — and spin up a separate query for each of those day-batches.

So what?

Some of you might be reading this and thinking, “I already do this with dbt, in [highly customized way].” That’s great! If it’s working for you, you don’t need to change a thing — we are committed to ongoing compatibility with all the code that’s written in existing dbt projects, including “classic” incremental models.

dbt is wonderfully flexible and extensible. At the same time, whenever we see dozens or hundreds of people modifying or extending dbt to accomplish the same basic goal — that’s a clue that there’s something missing from the dbt standard. If dbt Core can support time-aware filtering and batched incremental processing out of the box, we can spend less time solving the same problem in the same way (or slightly different ways) — not to mention, maintaining internal documentation for hand-rolled homegrown approaches — and more time solving the problems that are unique to our data. That’s what we mean by moving up the stack.

[rumbin]

This is certainly an important step into the right direction :-).

I'd like to add some thoughts to this discussion:

• The _etl_loaded_at field may not always be fully dependable in distributed systems. E.g., Snowflake's Snowpipe provides the START_SCAN_TIME metadata field, but there's no such thing as a data_arrived_in_lake_table_at field. Adding a constant incremental time overlap is, however, not always the most efficient solution.
• Late-arriving data may require a re-ingestion of all data of the entities included in the late-arriving batch, if, e.g., deduplication or aggregation along these entities is done in the model. E.g., consider Change Data Capture (CDC) data, of which a batch arrived out-of-order. So the latest batch (ETL time) is not the most recent state in terms of event time. It depends on the producer and data model logic whether we can just discard the late batch, or if we need to construct the current state out of all the batches of the affected entities.

In my daily work, these considerations are ubiquitous, as all our sources are async and typically at-least-once delivered. So, any generic solution which would not account for such conditions would require us to still stick with our custom incremental logic.

[jeff-skoldberg-gmds]

Thanks to @dataders for calling this thread to my attention.

I wanted to weigh in on the full refresh topic. I don't see the options as 3 separate options.

I think options 1 and 2 go together: we should allow full refresh (option 1) but the default is full_refresh=false (option 2). In order to accidently full refresh, they would need to commit a code change (config full_refresh=true), get that approved and merged, then run -f in the CLI.

As for option 3, cold storage, I think this is a necessary addition. But may consider disassociating it with the "full refresh" concept. Maybe a new flag is warranted that is something like purge_since or refresh_from_date. I hope you get what I mean without me explaining more :)

In summary:

I'd like to see very standard full-refresh behavior but dbt can take the liberty to make should_full_refresh default to false, and the user can change it. But we also need a way to "delete from table where date > x" then run.

[graciegoheen]

We're planning to add a new config for users to specific "when to start" when they are running in "full-refresh" mode (or for the initial build) - #10701

[jaypeedevlin]

We have a common scenario where an incremental model combines data from multiple sources which can have their own latency, eg we get event data throughout the day but data from a service every (for example) 6 hours. In these cases we generally want our max time to be calculated as follows:

1. Calculate the max time for each component model
2. Take the minimum of the maximums in 1

We would love this to be included functionality, otherwise incremental tables with left joins will having missing data and need to be rebuilt. Is this something you would consider?

[graciegoheen]

Your use case makes sense - our initial cut of this for 1.9 will likely just support using the current timestamp as the "max time" but this is something we should consider supporting in the future! How would you expect the configuration to look for something like this?

[seub]

Small question about the microbatching: will a single dbt run process several microbatches when needed? In other words, a single dbt build of the model will update the table several times? That's not quite clear to me from the description.

[graciegoheen]

Yes that's right - let's say your batch size is 1 day and you have 3 days of data to process. When you run that model, dbt will create 1 SQL query per batch to execute (so in this case 3).

We're doing some work now to support retrying only the failed batches -> #10729

[seub]

Got it thanks!

[dongchris]

How does this plan work with merge style tables, which is a separate incremental_strategy on its own?

[graciegoheen]

This won't affect existing incremental strategies! So you're current incremental models will continue to work as-is

[dongchris]

i guess the idea is that i would want to use some sort of --start-date --end-date natively (without needing to use vars) for incremental models (both insert overwrite & merge)

[datnguye]

Hey @graciegoheen I love this idea!

For now I am thinking with this new microbatch, if we can rerun the models with 1 or multiple batches in a specific period of event time for backfilling purpose or for fixing data after a bug-fixed?

e.g. dbt run -s +my_model —batch event_start_date=2024-08-15,event_end_date=2024-08-30

Thanks

[graciegoheen]

Yes! Our thinking is aligned :)

Our plan is to have CLI flags like you described to specify a time range for targeted backfilling:

dbt run --event-time-start "2024-09-01 00:00:00" --event-time-end "2024-09-04 00:00:00"

This specific time range would be applied to all of the microbatch models you've selected.

So in order to have different time ranges for different microbatch models, you'd need to run multiple dbt runs like:

dbt run -s my_fav_model --event-time-start "2024-09-01 00:00:00" --event-time-end "2024-09-04 00:00:00"
dbt run -s my_second_fav_model --event-time-start "2024-09-08 00:00:00" --event-time-end "2024-09-12 00:00:00"

[nravu-UOM]

Thank you @graciegoheen, keen to understand how this feature would work if my upstream model is a view and in my main model if I have to apply event_time filter on the upstream view. Can we process both the models in sequence in microbatch like in the traditional way of running the down-streams with + operator.
I also want to extend our custom incremental strategy with the microbatch feature, do you have any recommendations.

[alittlesliceoftom]

Hi @graciegoheen ,

This is great, and thanks @dataders for linking me to this from my discussion on incrementality here. Thanks so much for taking on this big topic!

Regarding the full refresh, from the experience of running with ibtp as a large number of our incremental models, I'd say it's really important to maintain a full refresh that looks and feels like an incremental full refresh. That is which you run, and after you run you have a replacement table. Until the final transaction switch your prod table shouldn't change.

e.g this option

dbt run -s my_microbatch_model --full-refresh will rebuild all batches for all time — similar to classic incremental models, although each batch will have its own query, and those queries will run independently.

Why is this important.

If you change some model in a fundamental way, you want to be able to rewrite it and all downstreams in one go. You shouldn't have to specify event time which might vary between models.

For us we set the lookback period quite short for dev so that you can dev fast, which I think is the same as 'sample' that you describe.

Consider supporting blue green deployments

Our main flaw at the moment is that the table starts by truncating , then filling row by row. Ideally this solution won't edit prod table until the replacement has been fully built, just like incremental models.

Given the size of those full refreshes, it would be nice to have a flag option to archive the old version of a full refreshed table, rather than just delete it.

p.s. apologies this message is a bit rushed!

[zmunro]

Will this incremantalize more complex DBT models that involve things like multiple joins and aggregations?
e.g.

select 
  c.id, 
  c.name, 
  count(se.customer_id) as num_events
from {{ ref('stg_events') }} se
join {{ ref('stg_customers') }} c on se.customer_id = c.id
group by c.id, c.name;

[graciegoheen]

Yes! Each ref/source statement will be turned into a "filtered ref/source" (assuming you've configured an event_time for the referenced model/source) and haven't opted out of auto-filtering.

[github-christophe-oudar]

Great initiative!
My feedback might be a bit long for Github so I wrote a whole article about it!
I'm looking forward to it 🙌

[anaghshineh]

First off – very excited to see this conversation! Thanks for putting so much great thought into this. Definitely guilty of this:

If you want to specify batches for your incremental model to process (”rebuild for 2024-09-01”), you need to add your own home-rolled logic, likely using vars

I want to +1 several of Christophe's points in his article. In particular:

• Instead of specifying microbatch as a new incremental_strategy, allow it to be an additional configuration of pre-existing incremental strategies. I find this more intuitive and extensible. When I think about building a model in batches, I'm thinking about what incremental strategy to apply to each batch, not of microbatching as its own incremental strategy.
• Use a more generic name than event_time. I have a lot of experience with large tables in BigQuery. I'll frequently want this to be the field the table is partitioned on. This does not always align with what I would think of as the event time.

[anaghshineh]

Oh, and an ergonomic way to run batches in a certain sequence would be awesome. I've had a painful time backfilling tables that require a sequential execution of batches. One way I've done this in the past is to use custom vars to specify batches and then set up an Airflow backfill pipeline that ensures these batches run in the proper sequence! Such a pain!

[graciegoheen]

Thank you so much for the thoughtful feedback here!

A few follow up questions I'd love to get your thoughts on:

an ergonomic way to run batches in a certain sequence would be awesome.

Would you imagine that this would be a new config to say "for this microbatch model I want my batches to always run in sequence"? Or are you thinking there'd be a way to infer that you had self-referencing logic?

The proposed ref('upstream_model').render() feels awkward. A cleaner approach might be an optional parameter in ref, like ref('upstream_model', micro_batch = false)

Is the thing that feels awkward here the name "render" - would something like ref('upstream_model').unfiltered() feel less awkward? Or is the awkward piece not having this within the ref itself?

The decision to add this feature as an incremental_strategy seems counterintuitive.

Could you say more about the use case where you'd want to process timeseries data in batches and merge on a unique key? The main use case we've been thinking about is massive timeseries data where you'd want to do "entire partition replacement" - the entire old batch is deleted and replaced by the new batch. Having microbatch as an incremental_strategy allows us to provide a further abstraction - you don't have to think "is delete + insert the best way to do this entire partition replacement?" - we can instead pursue whatever is the most efficient operation on the underlying DWH. Curious to hear more of your thoughts on this!

[github-christophe-oudar]

Is the thing that feels awkward here the name "render" - would something like ref('upstream_model').unfiltered() feel less awkward? Or is the awkward piece not having this within the ref itself?

I feel like the common pattern is {{ ref('my_model') }} having to call a function on it in some cases feels a bit weird to me. Render is mostly implicit and expected as is for most developers in my opinion.
I prefer to have a consistent approach and pack it in the ref function but maybe that's just a personal taste.
There could cases where it would make sense to make it evolve with different render method.

Could you say more about the use case where you'd want to process timeseries data in batches and merge on a unique key? The main use case we've been thinking about is massive timeseries data where you'd want to do "entire partition replacement" - the entire old batch is deleted and replaced by the new batch. Having microbatch as an incremental_strategy allows us to provide a further abstraction - you don't have to think "is delete + insert the best way to do this entire partition replacement?" - we can instead pursue whatever is the most efficient operation on the underlying DWH. Curious to hear more of your thoughts on this!

I was thinking about cases where you have a massive message broker with raw events that is dumped in the DWH, then you want to partition those events on not the time they landed in the broker or the DWH but on the event timestamp (that could be the time when the event happened on a device that was on flight mode). Then you're going to write append (or merge) on the event timestamp. You might have an unique key to know if the event exists or was duplicated but you dont want to explicitly rewrite a query that union the batch with the existing partitions and then achieve the X, Y, ANY_VALUE(...) group by X, Y

[anaghshineh]

Curious to hear more about your vision for how this plays out when running a DAG with at least one microbatch model dependent on another microbatch model. For example, let's take a simple daily pipeline with microbatch model A and microbatch model B downstream of it. A batch for each model is a day. If we were to backfill, say, a month, would dbt wait for all batches of A to complete before running B? Or would there be some smarter logic under the hood to, say, recognize that you can run a given day for model B once that same day has been processed for A?

And if B is not incrementally built, dbt would know to wait for A to complete all its batches before starting B?

[github-christophe-oudar]

Right, the difference of granularity amongst dependencies is often a problem to deal with. If the backfill is sequential you could have progressively partial rollups (for hourly to day, the daily rollup is filled with 1 to 24 hours) but it can be missleading for downstream consumers if they can't know "when" it's ready (and thus requiring yet another custom logic to manage that status)

[graciegoheen]

A* ----> B

*where A is a microbatch model

In what we've been working on so far, we wait to build model B until all of model A's batches have completed successfully (regardless of how model B is materialized). Maybe there's something smarter we can do in the future like you've suggested - "to, say, recognize that you can run a given day for model B once that same day has been processed for A" - but agree that differences in granularity could make that tricky!