Dagsterin käyttöönotto ja hyödyt dataputkien hallinnassa

Mikä Dagster on ja mitä sillä tehdään?

Dagster on dataputkien orkestrointityökalu. Orkestrointi tarkoittaa työn järjestämistä: mikä aineisto tarvitaan ensin, mikä laskenta riippuu siitä, milloin ajo käynnistyy ja miten tulos tarkistetaan. Dagster kokoaa näiden vaiheiden riippuvuudet ja ajohistorian yhteiseen käyttöliittymään.

Keskeinen käsite on asset eli tieto-omaisuus. Se voi olla esimerkiksi CRM:stä tuotu tilaustaulu, dbt:llä rakennettu myyntimalli tai ennusteen tulosaineisto. Assetin määrittely kertoo, miten aineisto muodostetaan ja mistä muista aineistoista se riippuu. Materialisointi tarkoittaa tämän määrittelyn suorittamista ja aineiston tuottamista.

BigQuery säilyttää ja käsittelee dataa, dbt määrittelee SQL-muunnoksia ja Dagster koordinoi ajoketjua. Dagsteriin voi liittää myös Pythonilla tehtäviä tiedonsiirtoja ja muita laskentoja. Riippuvuudet täytyy kuvata toteutuksessa: työkalu ei päättele kaikkia yrityksen järjestelmien välisiä yhteyksiä itsestään.

• Dagster: assetien määrittely ja riippuvuudet

Mitä hyötyä Dagsterista on yritykselle?

Hyöty syntyy erityisesti ylläpidossa. Kun aamun raportti ei ole päivittynyt, tiimi tarvitsee vastauksen siihen, mikä aineisto puuttuu, missä ajo pysähtyi ja mitä voidaan tehdä seuraavaksi. Yhteinen ajohistoria vähentää tarvetta etsiä tätä tietoa erillisistä skripteistä ja lokipalveluista.

• Dagster: käyttöliittymä, ajot ja riippuvuusnäkymät

• Selkeämpi tilannekuva: määritellyistä riippuvuuksista nähdään, mihin seuraaviin aineistoihin epäonnistunut vaihe vaikuttaa.
• Hallittu suoritusjärjestys: samassa ajossa mukana olevat riippuvat vaiheet voidaan suorittaa oikeassa järjestyksessä.
• Rajatut uudelleenajot: epäonnistunut vaihe tai valittu historiaväli voidaan käsitellä uudelleen, kun tallennus ja riippuvuudet tukevat sitä.
• Laadun seuranta: sovitut tarkistukset, kuten puuttuvat tunnisteet tai odottamaton rivimäärä, voidaan liittää aineiston käsittelyyn.
• Helpompi vastuun siirto: versionhallittu toteutus ja näkyvä ajohistoria auttavat myös muuta tiimiä ymmärtämään kokonaisuutta.

Mitattavia tavoitteita voivat olla häiriön selvittämiseen kulunut aika, käsin tehtävien uudelleenajojen määrä ja raportin valmistuminen sovittuun aikaan. Dagster ei sellaisenaan nopeuta SQL-kyselyä tai pienennä BigQuery-laskua. Säästö edellyttää esimerkiksi turhan käsittelyn vähentämistä tai datamallien optimointia.

Milloin käyttöönotto kannattaa ja milloin ajastus riittää?

Yksi päivittäinen tiedonsiirto voi toimia hyvin ajastetulla Python-skriptillä tai Cloud Run Jobilla, kun virheiden käsittely ja valvonta ovat kunnossa. Orkestraattorin käyttöönotolle kannattaa nimetä ongelma, jonka nykyinen ratkaisu jättää hoitamatta.

Dagsteria kannattaa arvioida, kun lähteet valmistuvat eri aikaan, ajoketjussa on useita riippuvuuksia tai historiatietoja pitää korjata toistuvasti. Myös usean henkilön ylläpitämä kokonaisuus hyötyy siitä, että suoritusten tila ja riippuvuudet ovat kaikkien nähtävissä.

Arvioikaa samalla käyttöönoton ja ylläpidon työ. Pienen ajoketjun siirtäminen uuteen järjestelmään ei ole hyödyllistä, jos ongelma ratkeaisi yhdellä lähdetarkistuksella ja hälytyksellä. Aloittakaa yhdestä putkesta ja verratkaa sen ylläpitoa nykyiseen toimintatapaan.

• Kevyempi lähtökohta: rajapintasiirto Pythonilla ja Cloud Runilla

Rajaa ensimmäinen Dagster-toteutus

Valitkaa ensimmäiseksi tuttu ajoketju, jonka lopputuloksen pystytte tarkistamaan. Sopiva pilotti voisi olla tilaustietojen tuonti, päiväkohtainen myyntiyhteenveto ja sen laadun tarkistus. Esimerkki kuvaa mahdollista toteutusta, ei yksittäistä asiakasprojektia.

Kirjatkaa lähteen saatavuus, aineiston päivitystavoite ja vastuuhenkilö. Päättäkää myös, millä perusteella pilotti hyväksytään: esimerkiksi oikeat luvut, onnistunut uudelleenajo ja näkyvä ilmoitus tarkoituksella aiheutetusta virheestä.

• Nimetkää lähdeaineisto ja raportointiin tuotettava lopputulos.
• Kuvatkaa riippuvuudet ja päättäkää, missä data säilytetään eri vaiheiden välillä.
• Sopikaa, miten sama aineisto käsitellään uudelleen ilman kaksoiskappaleita.
• Valitkaa tarkistukset ja määritelkää, kuka reagoi niiden epäonnistumiseen.

1. Luo paikallinen Dagster-projekti

Tarvitset Python 3.10:n tai uudemman sekä uv-paketinhallinnan. Seuraava kokeilu käyttää vain itse määriteltyä esimerkkidataa. Pilvitiliä, BigQuery-yhteyttä tai tuotannon tunnuksia ei tarvita. Luo projekti hakemistoon, jossa voit tehdä kehitystyötä.

Suorita ensimmäinen komento ja vastaa myöntävästi uv sync -kysymykseen. Siirry sitten projektihakemistoon. Komennot käyttävät uv run -muotoa, joten virtuaaliympäristöä ei tarvitse aktivoida erillisellä komennolla.

• uv: asennusohje eri käyttöjärjestelmille
• Dagster: asennuksen edellytykset

uvx create-dagster@latest project dagster-aloitus
cd dagster-aloitus
uv sync
uv run dg --version

Projektipohja sisältää pyproject.toml-tiedoston ja Python-paketin src/dagster_aloitus. Määrittelyt sijoitetaan sen defs-hakemistoon. Säilytä projektipohjan definitions.py: se lataa tämän hakemiston määrittelyt. Vie lähdekoodi ja uv.lock versionhallintaan, jotta tiimi käyttää samoja riippuvuusversioita.

Oppaan komennot ja Python-esimerkki on tarkistettu Dagsterin versiolla 1.13.24 ja Pythonin versiolla 3.12.3. Uusin projektipohja voi asentaa myöhemmin eri version.

2. Määrittele aineistot ja niiden riippuvuus

Luo tiedosto src/dagster_aloitus/defs/myynti.py ja lisää siihen alla oleva koodi. Ensimmäinen asset tuottaa kolme esimerkkitilausta. Toinen laskee niiden lukumäärän ja myyntisumman. Rahamäärät käsitellään kokonaisina sentteinä.

Myyntiyhteenvedon tilaukset-parametri määrittelee riippuvuuden samannimiseen assetiin. Kun valitset molemmat suoritettaviksi, Dagster muodostaa ensin tilaukset ja välittää tuloksen yhteenvedolle. Metatiedot auttavat tarkistamaan tuloksen käyttöliittymässä.

• Dagster: datan välittäminen assetien välillä

import dagster as dg


@dg.asset
def tilaukset() -> list[dict]:
    return [
        {"tilaus_id": "demo-1", "summa_sentteina": 12000},
        {"tilaus_id": "demo-2", "summa_sentteina": 3500},
        {"tilaus_id": "demo-3", "summa_sentteina": 4500},
    ]


@dg.asset
def myyntiyhteenveto(
    context: dg.AssetExecutionContext,
    tilaukset: list[dict],
) -> dict:
    if not tilaukset:
        raise dg.Failure("Esimerkin tilausaineisto on tyhjä.")

    yhteenveto = {
        "tilauksia": len(tilaukset),
        "myynti_sentteina": sum(
            rivi["summa_sentteina"] for rivi in tilaukset
        ),
    }
    context.add_output_metadata(yhteenveto)
    return yhteenveto

Esimerkki käyttää Dagsterin oletusarvoista paikallista tulosten tallennusta. Tuotannossa aineistot voidaan kirjoittaa esimerkiksi BigQueryyn tai yhteiseen objektitallennukseen. Prosessien välillä tarvitaan kaikille tarvittaville suorituksille saavutettava tallennus; yhden kehityskoneen tiedostot eivät riitä hajautettuun toteutukseen.

3. Tarkista määrittelyt ja aja esimerkki

Suorita komennot projektin juuresta. Ensimmäinen tarkistaa, että määrittelyt latautuvat. Toinen ajaa kaikki esimerkin assetit. Viimeinen käynnistää paikallisen kehitysympäristön, jonka käyttöliittymä avautuu oletuksena osoitteessa http://localhost:3000.

• Dagster: ensimmäisen putken suorittaminen ja käyttöliittymä

uv run dg check defs
uv run dg launch --assets "*"
uv run dg dev

Avaa käyttöliittymän Assets-näkymä, valitse molemmat assetit ja käynnistä materialisointi. Riippuvuusgraafissa näkyy tilaukset → myyntiyhteenveto. Tarkista ajon onnistuminen ja yhteenvedon metatiedot: tilauksia on 3 ja myynti_sentteina on 20000 eli 200 euroa. Komentorivin kerta-ajo ja kehitysympäristö voivat käyttää eri tilapäistä ajohistorian tallennusta, joten aja esimerkki myös käyttöliittymästä.

Kokeile virhetilannetta muuttamalla tilaukset-funktion palauttama lista tyhjäksi. Tallenna tiedosto, lataa määrittelyt uudelleen käyttöliittymän Reload definitions -toiminnolla ja materialisoi molemmat assetit. Yhteenveto epäonnistuu ja näyttää määritellyn virheviestin. Palauta lopuksi esimerkkitilaukset, tallenna ja lataa määrittelyt uudelleen. Oikeassa toteutuksessa sovitaan erikseen, onko tyhjä lähdeaineisto virhe vai hyväksyttävä tilanne.

Kehitysympäristön voi pysäyttää Ctrl+C:llä. dg dev on paikalliseen kehitykseen; tuotannossa prosessien valvonta ja tilatiedon pysyvyys järjestetään erikseen.

Ajastukset, sensorit ja laadun tarkistukset

Lisää automaatio vasta, kun käsin käynnistetty ajo ja virhetilanne toimivat. Ajastus käynnistää työn sovittuun aikaan. Sensori tarkistaa esimerkiksi tiedoston saapumista tai muuta määriteltyä ehtoa ja voi pyytää ajon sen täytyttyä. Lähdeaineiston valmius täytyy todeta kummassakin vaihtoehdossa.

Määrittele ajastukselle aikavyöhyke ja huomioi kesäajan vaikutus. Sensorissa ajopyynnön tunniste, run_key, voi estää saman tapahtuman toistuvat ajopyynnöt. Se ei kuitenkaan poista kaksoiskappaleita kohdedatasta tai korvaa epäonnistuneiden suoritusten uusintakäytäntöä.

Asset check -tarkistus voi valvoa esimerkiksi tunnisteiden yksikäsitteisyyttä tai tietojen tuoreutta. Kaikki tarkistukset eivät pysäytä myöhempää laskentaa: estävä tarkistus määritellään erikseen. Sovita hälytykset käytössä olevaan Dagster-versioon ja ympäristöön, ja sovi myös kuka ne vastaanottaa.

• Dagster: ajastukset
• Dagster: sensorit ja ajopyynnöt
• Dagster: laadun tarkistukset

Partitiot ja historian uudelleenajo

Kun aineisto kertyy esimerkiksi päivittäin, asset voidaan jakaa päiväkohtaisiin partitioihin. Dagsterin partitio kertoo, mitä osaa aineistosta ajo käsittelee. Se ei automaattisesti luo BigQuery-taulun osiointia: tallennuksen rakenne ja suodatus toteutetaan erikseen.

Backfill tarkoittaa valitun historiavälin käsittelyä jälkikäteen. Se on hyödyllinen esimerkiksi laskentasäännön korjauksessa. Rajaa ensin pieni aikaväli, tarkista tulos ja hallitse rinnakkaisten ajojen määrää ennen laajaa uudelleenlaskentaa.

Uudelleenajon täytyy tuottaa sovittu lopputulos myös silloin, kun kyseinen päivä on jo käsitelty. Kohteessa voidaan käyttää esimerkiksi avaimiin perustuvaa päivitystä tai rajatun aineiston korvaamista. Pelkkä rivien lisääminen jokaisella ajolla voi monistaa tiedot. Myöhässä saapuvan datan tunnistus ja korjausikkuna suunnitellaan erikseen.

• Dagster: partitioidun aineiston backfill-ajot
• Syventävä opas: Dagster, GA4 ja dbt samassa dataputkessa

Mitä Dagsterin tuotantokäyttö edellyttää?

Valitse ylläpitomalli ennen pilotin laajentamista. Avoimen lähdekoodin Dagsterissa organisaatio vastaa ympäristöstä itse. Dagster+ tarjoaa hallittuja vaihtoehtoja: Serverlessissä myös koodin suoritus on palveluntarjoajan hallinnassa, Hybridissä koodi suoritetaan omassa ympäristössä agentin kautta. Tarkista vaihtoehtojen ajantasaiset ominaisuudet, hinnoittelu ja tietojen käsittely ennen valintaa.

Itse ylläpidetty kokonaisuus tarvitsee käyttöliittymää palvelevan webserverin, koodin latauksen ja suorituksen sekä ajotietojen tallennuksen. Ajastukset ja sensorit tarvitsevat toimivan daemon-prosessin. Tapahtumat ja ajohistoria on säilytettävä pysyvästi, esimerkiksi tuotantoon soveltuvassa PostgreSQL-tietokannassa.

• Dagster OSS: tuotantoympäristön osat
• Dagster+: hallitut käyttöönottovaihtoehdot

• Erota kehitys ja tuotanto sekä rajaa käyttöoikeudet lähteisiin ja kohteisiin. Säilytä tunnukset ympäristön salaisuuksien hallinnassa.
• Järjestä pysyvä tilatieto, tarvittava yhteinen aineistotallennus ja varmuuskopioiden palautuksen kokeilu.
• Valvo daemonin, koodipalvelujen ja suoritusten toimintaa. Testaa ilmoitus myös tilanteessa, jossa ajo ei käynnisty lainkaan.
• Rajoita yhtäaikaisia suorituksia lähdejärjestelmien kapasiteetin ja kustannusten perusteella.
• Sovi ohjelmistopäivitykset, julkaisukäytäntö, häiriöiden vastuuhenkilöt ja dokumentaation ylläpito.

Dagsterin käyttöliittymään tallentuu lokeja ja metadataa, joten myös niiden sisältö ja käyttöoikeudet on suunniteltava. Tietokannan ajohistoria ja varsinaisen liiketoimintadatan tallennus ovat eri asioita. Kustannuksiin kuuluvat valitun palvelun lisäksi infrastruktuuri, datan käsittely ja ylläpitotyö.

Miten Dagster liittyy BigQueryyn ja dbt:hen?

Tyypillisessä kokonaisuudessa Python-siirto tuo lähdedatan BigQueryyn, dbt muodostaa raportointimallit ja Data Studio näyttää sovitut mittarit. Dagster voi koordinoida tiedonsiirron, mallien ajon ja laadun tarkistukset. Raportin oman välimuistin ja päivitysten toiminta sovitaan erikseen.

dagster-dbt-integraatio tuo dbt-mallien riippuvuudet osaksi Dagsterin asset-mallia. BigQuery-yhteys määritellään resurssina ja sen käyttöoikeudet rajataan työn tarpeeseen. Aloita olemassa olevista malleista ja tiedonsiirroista: kaikkia laskentoja ei tarvitse kirjoittaa uudelleen käyttöönoton vuoksi.

• Dagster: dbt-integraatio
• Dagster: BigQuery-integraatio
• GA4-datan mallinnus BigQueryssa dbt:n avulla

Miten arvioida käyttöönoton onnistumista?

Vertaa pilotin tuloksia ennen käyttöönottoa kirjattuun lähtötilanteeseen. Oikein päivittyvä riippuvuusgraafi on hyödyllinen, mutta käyttöönoton arvo näkyy siinä, miten luotettavasti aineisto valmistuu ja miten tiimi pystyy hoitamaan poikkeamat.

Kokeile onnistuneen ajon lisäksi puuttuvaa lähdeaineistoa, epäonnistunutta työvaihetta ja historian korjausta. Varmista, että toinen ylläpitäjä osaa löytää virheen ja tehdä sovitun uudelleenajon dokumentaation perusteella. Laajenna kokonaisuutta vasta, kun vastuut ja toimintatavat ovat selvät.

• Valmistuuko aineisto sovittuun aikaan ja ovatko sen luvut oikein?
• Löytyykö häiriön syy aiempaa helpommin?
• Onnistuuko uudelleenajo ilman kaksoiskappaleita ja tarpeetonta koko historian käsittelyä?
• Vastaavatko ylläpidon työ ja kustannukset saavutettua hyötyä?