Dagster ja dbt: GA4-dataputken orkestrointi

Mitä Dagsterilla tehdään

Dagster on dataputkien orkestrointityökalu. Sen avulla kuvataan työvaiheiden riippuvuudet, käynnistetään ajot ja seurataan niiden tuloksia. Se täydentää tiedonsiirtoja ja dbt-mallinnusta yhteisellä hallinnalla.

Dagsterin asset-lähtöisessä mallissa kuvataan tietoaineistoja, kuten tauluja, ja niiden syntytapa. Riippuvuusnäkymä auttaa selvittämään, mistä raportin tieto tulee ja mikä työvaihe tarvitsee huomiota. Myös muissa orkestraattoreissa on vastaaviin tarpeisiin soveltuvia ominaisuuksia.

• Aloita tästä: Dagsterin käyttöönotto ja hyödyt

Tarvitsetteko Dagsteria?

Yksittäinen dbt-ajo voi toimia Cloud Run Jobsin ja Cloud Schedulerin avulla, kun lähdetarkistukset, uusinnat ja valvonta on toteutettu. Orkestraattoria kannattaa arvioida silloin, kun näiden asioiden hallinta erillisillä ajastuksilla käy vaikeaksi.

Dagster voi olla perusteltu esimerkiksi seuraavissa tilanteissa:

• dbt-mallien ajaminen Cloud Runissa
• Yksittäisestä rajapintasiirrosta orkestroituun dataputkeen

• Lähteitä on useita ja ne valmistuvat eri aikaan: esimerkiksi GA4-vienti, CRM-poiminta ja mainoskulut.
• Osien välillä on aitoja riippuvuuksia: aggregaattia ei ole mitään järkeä laskea ennen kuin kaikki sen lähteet ovat kunnossa.
• Historiaa joudutaan laskemaan uudelleen: määritelmä muuttui, ja kuusi kuukautta pitää ajaa läpi hallitusti.
• Joku muu kuin putken rakentaja tarvitsee näkymän siihen, mikä on ajan tasalla ja mikä ei.
• Vikatilanteessa pitää tietää mikä osa epäonnistui ja voiko sen ajaa uudelleen yksin, ilman että koko yö ajetaan alusta.

Jos ajoketju on yksinkertainen, aloittakaa sen vaatimuksista ja verratkaa orkestroinnin lisähyötyä ylläpidon kustannukseen. dbt-oppaassa on esimerkki Cloud Runissa ajettavasta toteutuksesta.

GA4-tapahtumat partitioituna assettina

GA4:n vienti on päiväkohtainen: yksi taulu per vuorokausi. Se sopii Dagsterin partitiointimalliin lähes täydellisesti. Määrittelet assetin, jolla on päiväpartitiot, ja jokainen partitio vastaa yhtä GA4-shardia.

Nimetyn päiväpartition voi ajaa uudelleen rajatusti, ja ajohistoriasta nähdään sen tila. Myöhäiset muutokset vaativat silti erillisen tunnistuksen tai liukuvan uudelleenlaskentaikkunan; partitiointi ei havaitse niitä itsestään.

from dagster import asset, AssetExecutionContext, DailyPartitionsDefinition
from dagster_gcp import BigQueryResource

daily = DailyPartitionsDefinition(
    start_date="2025-01-01", timezone="Europe/Helsinki"
)


@asset(partitions_def=daily, group_name="ga4", kinds={"bigquery"})
def stg_ga4_events(
    context: AssetExecutionContext,
    bigquery: BigQueryResource,
) -> None:
    """GA4:n raakatapahtumat litteäksi tauluksi, yksi päivä kerrallaan."""
    day = context.partition_key              # "2025-07-01"
    shard = day.replace("-", "")             # "20250701"

    sql = f"""
    begin transaction;

    delete from `analytics.stg_ga4_events`
     where event_date = date '{day}';

    insert into `analytics.stg_ga4_events`
      (event_date, event_at, event_name, user_pseudo_id, user_id,
       ga_session_id, page_location, source, medium,
       transaction_id, purchase_revenue)
    select
      date '{day}'                                  as event_date,
      timestamp_micros(event_timestamp)             as event_at,
      event_name,
      user_pseudo_id,
      user_id,
      (select value.int_value from unnest(event_params)
         where key = 'ga_session_id')               as ga_session_id,
      (select value.string_value from unnest(event_params)
         where key = 'page_location')               as page_location,
      collected_traffic_source.manual_source        as source,
      collected_traffic_source.manual_medium        as medium,
      ecommerce.transaction_id,
      ecommerce.purchase_revenue
    from `projekti.analytics_123456789.events_{shard}`;

    commit transaction;
    """

    with bigquery.get_client() as client:
        job = client.query(sql)
        job.result()

    context.add_output_metadata({"shard": shard, "bytes_billed": job.total_bytes_billed})

Luo kohdedatasetti ja event_date-kentällä partitioitu kohdetaulu ennen ajoa, yllä olevan sarakeluettelon mukaisesti. Suora päivätauluviittaus tuottaa virheen, jos lähde puuttuu, jolloin transaktio ei poista vanhaa dataa. Delete ja insert samassa transaktiossa tekevät uudelleenajosta idempotentin, kun lähde on sama.

Kirjatkaa ajon käsittelemä datamäärä ja keskeiset tulostiedot metadataan. Näin poikkeavia ajoja voidaan verrata aikaisempiin ja tarvittaessa yhdistää BigQueryn tarkempaan kustannusseurantaan.

Miten dbt liittyy Dagsteriin

Tämä on se kohta, jossa yhdistelmä alkaa kannattaa. dagster-dbt-kirjasto lukee dbt-projektin manifestin ja luo jokaisesta dbt-mallista oman Dagster-assettinsa. dbt:n sisäinen riippuvuusgraafi ei siis jää dbt:n sisään, vaan sulautuu samaan graafiin muun putken kanssa.

Yhteinen riippuvuusnäkymä yhdistää lähdepoiminnat, dbt-mallit ja raportointitaulut. Se helpottaa häiriön rajaamista, mutta lukujen oikeellisuus vaatii edelleen sisältötarkistuksia ja ymmärrystä laskentasäännöistä.

import json
from pathlib import Path

from dagster import AssetExecutionContext, AssetKey
from dagster_dbt import (
    DagsterDbtTranslator,
    DbtCliResource,
    DbtProject,
    dbt_assets,
)

from .ga4 import daily

dbt_project = DbtProject(
    project_dir=Path(__file__).joinpath("..", "..", "dbt_projekti").resolve()
)
dbt_project.prepare_if_dev()


class Ga4Translator(DagsterDbtTranslator):
    """Nimeää DBT:n lähteet samoiksi asseteiksi kuin Python-puolen assetit.

    Ilman tätä DBT:n source 'analytics.stg_ga4_events' saisi eri asset-avaimen
    kuin yllä määritelty stg_ga4_events, ja graafi katkeaisi juuri siitä
    kohdasta, jonka takia se rakennettiin.
    """

    def get_asset_key(self, dbt_resource_props):
        if dbt_resource_props["resource_type"] == "source":
            return AssetKey(dbt_resource_props["name"])
        return super().get_asset_key(dbt_resource_props)


@dbt_assets(
    manifest=dbt_project.manifest_path,
    dagster_dbt_translator=Ga4Translator(),
    partitions_def=daily,
)
def dbt_analytics_assets(context: AssetExecutionContext, dbt: DbtCliResource):
    # Mallien on käytettävä tätä päivää lähderajauksessa ja korvattavissa partitioissa.
    dbt_vars = {"partition_date": context.partition_key}
    yield from dbt.cli(
        ["build", "--vars", json.dumps(dbt_vars)], context=context
    ).stream()

dbt build ajaa valitut mallit ja testit. dagster-dbt-integraatio tuo niiden suoritustiedot samaan näkymään muun ajoketjun kanssa.

Dagster-partitio ei itsestään rajaa dbt:n SQL:ää. Päivämallien on käytettävä var("partition_date")-arvoa ja korvattava vastaava BigQuery-partitio, myös tyhjällä tuloksella. Istunnot voivat ylittää vuorokauden rajan: lue niiden rakentamiseen tarvittavat viereiset päivät ja kohdista tulos istunnon alkupäivälle. dbt-oppaan koko historian taulumallit eivät sellaisenaan toteuta tätä partitioitua ajoa.

Datan yhdistäminen muista lähteistä

Jokainen lähde on oma assettinsa omalla aikataulullaan. Mainoskulut noudetaan rajapinnasta, CRM-data tietokannasta, budjetit ehkä Sheetsistä. Dagster ei vaadi niitä samaan tahtiin. Se vaatii vain, että riippuvuudet on kerrottu.

Riippuvuus kuvataan parametrina tai deps-määrittelyssä. Se määrää samassa ajossa valittujen assettien järjestyksen, mutta ei yksin varmista lähteiden tuoreutta tai käynnistä niitä. Tarvitaan myös oikea ajovalinta sekä saatavuus- ja laatutarkistukset.

from dagster import AssetExecutionContext, asset
from dagster_gcp import BigQueryResource
from .ga4 import daily


@asset(partitions_def=daily, group_name="ads", kinds={"bigquery"})
def stg_google_ads_cost(
    context: AssetExecutionContext,
    bigquery: BigQueryResource,
) -> None:
    """Mainoskulut päivätasolla. Oma lähde, oma viive."""
    raise NotImplementedError("Toteuta mainoskulujen nouto ja saatavuustarkistus")


@asset(partitions_def=daily, group_name="crm", kinds={"bigquery"})
def stg_crm_orders(
    context: AssetExecutionContext,
    bigquery: BigQueryResource,
) -> None:
    """Tilaukset CRM:stä. Sisältää peruutukset ja katteen, joita GA4 ei tiedä."""
    raise NotImplementedError("Toteuta CRM-poiminta ja saatavuustarkistus")

-- Yksi rivi per päivä ja kanava ennen liitosta.
-- fct_orders_attributed: yksi rivi per tilaus, ks. DBT-opas.
with sessions as (
    select session_date as date, channel_group, count(*) as sessions
    from {{ ref('fct_sessions') }}
    group by 1, 2
),
orders as (
    select date(ordered_at, 'Europe/Helsinki') as date,
           channel_group, count(*) as orders, sum(margin) as margin
    from {{ ref('fct_orders_attributed') }}
    where not is_cancelled
    group by 1, 2
),
cost as (
    select date, channel_group, sum(ad_cost) as ad_cost
    from {{ source('analytics', 'stg_google_ads_cost') }}
    group by 1, 2
)
select
  coalesce(s.date, o.date, c.date) as date,
  coalesce(s.channel_group, o.channel_group, c.channel_group) as channel_group,
  coalesce(s.sessions, 0) as sessions,
  coalesce(o.orders, 0) as orders,
  coalesce(o.margin, 0) as margin,
  coalesce(c.ad_cost, 0) as ad_cost,
  safe_divide(c.ad_cost, o.margin) as cost_margin_ratio
from sessions s
full join orders o
  on s.date = o.date and s.channel_group is not distinct from o.channel_group
full join cost c
  on coalesce(s.date, o.date) = c.date
 and coalesce(s.channel_group, o.channel_group) is not distinct from c.channel_group

Esimerkki näyttää liitoslogiikan. Tee siitä partitioitu malli edellä kuvatulla päivärajaus- ja korvausmenetelmällä. Aggregointi ennen liitoksia estää istuntojen ja katteen monistumisen; täysliitokset säilyttävät myös päivät, joilla on pelkkiä kuluja tai tunnistamattomia tilauksia. Lähdepoiminnan on epäonnistuttava puuttuessaan, jotta puuttuvaa aineistoa ei tulkita nollaksi.

Aggregointi Data Studiota varten

Aggregointi itsessään tehdään dbt:llä täsmälleen kuten ilman Dagsteria: valmiiksi lasketut päivätaulut, partitiointi päivämäärän mukaan, klusterointi yleisimmän suodattimen mukaan. Raportti lukee tuhansia rivejä miljoonien sijaan.

Dagster näyttää riippuvuudet ja ajohistorian. Se ei takaa raportin atomista päivittymistä usean taulun yli: julkaise tarkistettu raportointiversio erikseen, jos lukijan pitää nähdä vain kokonaan valmis aineisto.

Valitkaa aggregaattien tarkkuustaso raportin kysymysten ja suodattimien perusteella. Useampi tarkoitukseen rajattu taulu voi helpottaa käyttöä, mutta myös niiden ajot ja ylläpito maksavat. Vertaile vaihtoehtoja todellisilla kyselyillä.

Missä Dagster ja dbt ajetaan

Itse ylläpidetyssä Dagsterissa sensorit ja ajastukset tarvitsevat jatkuvasti toimivan daemon-prosessin. Lisäksi järjestetään käyttöliittymä, koodin suoritus ja ajotietojen säilytys. Tuotantoympäristössä tilatietoon käytetään usein PostgreSQL:ää; paikalliskäytössä on myös kevyempiä vaihtoehtoja.

Toteutustapa valitaan ylläpito-osaamisen, ympäristön ja hallintavaatimusten perusteella. Esimerkkejä vaihtoehdoista:

• Dagster+ Hybridin arkkitehtuuri ja ylläpitovastuut

• Dagster+ Hybrid: hallintataso toimii Dagsterin palvelussa ja käyttäjän koodi omassa ympäristössä agentin kautta. Palveluun välitetään orkestroinnin tietoja, ja lokien sekä metadatan käsittely on tarkistettava asetuksista. Agentin ja suoritusympäristön ylläpito jää organisaatiolle.
• Avoimen lähdekoodin Dagster omassa ympäristössä, esimerkiksi GKE:ssä: infrastruktuuri, tietokanta, päivitykset ja valvonta ylläpidetään itse. Vaihtoehto voi sopia jo käytössä olevaan Kubernetes-ympäristöön.
• Itse hostattu Cloud Runissa: mahdollista, mutta daemon tarvitsee jatkuvan CPU-varauksen (instance-based billing) ja vähintään yhden instanssin. Pelkkä min-instances ei varmista taustaprosessin CPU-aikaa. Silloin maksatte jatkuvasti käynnissä olevasta instanssista, ja suuri osa Cloud Runin kustannusedusta katoaa.

from dagster import Definitions, define_asset_job, AssetSelection
from dagster_dbt import DbtCliResource
from dagster_gcp import BigQueryResource

from .assets.ga4 import daily, stg_ga4_events
from .assets.sources import stg_google_ads_cost, stg_crm_orders
from .assets.dbt import dbt_analytics_assets, dbt_project
from .checks import ga4_event_count_matches_raw
from .sensors import ga4_export_sensor

ga4_daily_job = define_asset_job(
    name="ga4_daily",
    selection=AssetSelection.all(),
    partitions_def=daily,
)

defs = Definitions(
    assets=[
        stg_ga4_events,
        stg_google_ads_cost,
        stg_crm_orders,
        dbt_analytics_assets,
    ],
    asset_checks=[ga4_event_count_matches_raw],
    jobs=[ga4_daily_job],
    sensors=[ga4_export_sensor],
    resources={
        "bigquery": BigQueryResource(project="projekti"),
        "dbt": DbtCliResource(project_dir=dbt_project),
    },
)

Rajataan agentin, koodin ja suoritusten käyttöoikeudet niiden tehtäviin. Suositaan ympäristön palvelutilejä tai tuettua identiteettifederaatiota ja vältetään pitkäikäisiä avaintiedostoja.

Milloin käytetään sensoria ja milloin ajastusta?

Ajo voidaan käynnistää ajastuksella ja lähdetarkistuksella tai sensorilla, joka havaitsee tarvittavan ehdon täyttymisen. Sensori on hyödyllinen, kun esimerkiksi GA4-viennin saatavuus vaihtelee päivästä toiseen.

Seuraava sensori tarkistaa edellisen päivän taulun olemassaolon ja pyytää ajon sen löytyessä. Tuotantototeutuksessa tarvitaan myös aikaraja odotukselle, hälytykset ja epäonnistuneen ajon uusintatapa.

from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

from dagster import RunRequest, SkipReason, sensor
from dagster_gcp import BigQueryResource


@sensor(job_name="ga4_daily", minimum_interval_seconds=1800)
def ga4_export_sensor(context, bigquery: BigQueryResource):
    """Käynnistää ajon vasta kun eilisen GA4-shard on olemassa."""
    day = datetime.now(ZoneInfo("Europe/Helsinki")).date() - timedelta(days=1)
    shard = day.strftime("%Y%m%d")

    sql = f"""
    select 1
      from `projekti.analytics_123456789.INFORMATION_SCHEMA.TABLES`
     where table_name = 'events_{shard}'
    """
    with bigquery.get_client() as client:
        found = list(client.query(sql).result())

    if not found:
        return SkipReason(f"GA4-vientiä {shard} ei ole vielä saapunut.")

    # run_key tekee tästä idempotentin: sama päivä käynnistyy vain kerran,
    # vaikka sensori ehtisi nähdä taulun useasti.
    return RunRequest(run_key=f"ga4-{shard}", partition_key=day.isoformat())

Esimerkin run_key estää saman päivän uuden ajopyynnön myös epäonnistuneen ajon jälkeen; määritä uudelleenyritykset erikseen. Sensori tarkistaa vain eilisen: tuotannossa seuraa myös puuttuvia aiempia päiviä ja myöhäisiä päivityksiä. Taulun olemassaolo ei tarkoita, että päivän data olisi lopullinen. Käytä omaisuuden todellista aikavyöhykettä sensorissa ja partitioissa.

Ajastettua ajoa kannattaa silti pitää sensorin rinnalla varmistuksena niille malleille, jotka eivät riipu GA4:stä. Sensori vastaa siitä, milloin GA4-haara alkaa; ajastus siitä, ettei muu putki jää kiinni yhdestä myöhästyvästä lähteestä.

Backfill: historian hallittu uudelleenlaskenta

Määritelmän korjaus tai lähdedatan täydentyminen voi edellyttää historian uudelleenlaskentaa. Dagsterin partitiointi ja backfill-toiminnot tarjoavat tähän näkyvän ajohistorian ja mahdollisuuden rajata käsiteltävät päivät.

Partitioidussa assetissa sama asia on backfill: valitset partitiovälin, Dagster ajaa ne, ja käyttöliittymässä näkyy mitkä ovat valmiit, mikä on kesken ja mikä epäonnistui. Epäonnistuneet voi ajaa uudelleen yksin koskematta onnistuneisiin.

Myöhässä saapuva data tarvitsee uudelleenajon myös Dagsterissa. Partitiointi ei tunnista GA4:n korjauksia automaattisesti: toteuta muutosten seuranta tai ajasta liukuva uudelleenlaskentaikkuna. Edellisen esimerkin kerran päivässä laukeava sensori ei yksin hoida tätä.

Asset checkit: laadun valvonta samassa graafissa

Sekä dbt-testeillä että Dagsterin asset checkeillä voi verrata lähteen ja lopputuloksen rivimääriä. Valitse toteutus putken rakenteen mukaan; Dagster-check tuo tarkistuksen saman graafin osaksi myös Python-poiminnalle.

from dagster import AssetCheckResult, asset_check
from dagster_gcp import BigQueryResource
from .assets.ga4 import daily, stg_ga4_events


@asset_check(asset=stg_ga4_events, blocking=True, partitions_def=daily)
def ga4_event_count_matches_raw(context, bigquery: BigQueryResource):
    """Tarkistaa ajettavan päivän, myös historian uudelleenajossa."""
    day = context.partition_key
    shard = day.replace("-", "")
    sql = f"""
    select
      (select count(*)
         from `projekti.analytics_123456789.events_{shard}`) as raw_n,
      (select count(*) from `analytics.stg_ga4_events`
        where event_date = date '{day}') as modelled_n
    """
    with bigquery.get_client() as client:
        row = next(iter(client.query(sql).result()))

    return AssetCheckResult(
        passed=row.raw_n == row.modelled_n,
        metadata={"raakadata": row.raw_n, "mallinnettu": row.modelled_n},
    )

blocking=True estää samassa ajossa mukana olevat riippuvat assetit, kun tarkistus epäonnistuu ERROR-tasolla. Sopikaa myös, mitä raportin käyttäjä näkee häiriön aikana. Jos käyttöön jätetään aiempi aineisto, sen päivitysaika ja viiveen syy tulee näyttää selkeästi.

Yleisimmät virheet

• Orkestraattorin ylläpitovastuut otetaan käyttöön arvioimatta ensin, mitä ongelmia se ratkaisee.
• Historiaa käsittelevistä asseteista puuttuu tapa rajata ja uusia ajot turvallisesti.
• dbt ajetaan yhtenä läpinäkymättömänä komentona dagster-dbt-integraation sijaan. Graafiin jää musta laatikko juuri siihen kohtaan, jossa suurin osa logiikasta on.
• Sensori ilman run_keytä: sama ajo käynnistyy uudelleen joka herätyksellä.
• Ajo ei ole idempotentti. Partitio ajetaan uudelleen ja rivit kahdentuvat, jolloin uudelleenajo, koko mallin tärkein ominaisuus, muuttuu riskiksi.
• Daemonin unohtaminen itse hostatussa ympäristössä. Ilman jatkuvasti käynnissä olevaa daemonia sensorit ja ajastukset eivät laukea, ja putki näyttää toimivalta koska mikään ei kaadu.
• Asset checkit jätetään pois, koska dbt:ssä on jo testit. Lähteen ja lopputuloksen täsmäytys jää tekemättä. Sen voi toteuttaa joko dbt-testinä tai Dagster-checkinä.

Versioista

Dagsterin rajapinta on kehittynyt nopeasti, ja tämän oppaan esimerkit nojaavat vakiintuneisiin osiin: @asset, partitiot, dagster-dbt ja sensorit. Uudempia tapoja kuvata automaatiota ja projektirakennetta on tullut lisää, ja ne muuttuvat edelleen.

Käytännön suositus: kiinnitä versiot ja lue oman versiosi dokumentaatio ennen kuin kopioit esimerkkejä mistään, tämä opas mukaan lukien. Dagster-projektissa versionsa kertova requirements-tiedosto säästää enemmän aikaa kuin mikään yksittäinen ominaisuus.