Microsoft Fabric: Pipeline vs Notebook

Microsoftin Fabric tarjoaa pari erilaista vaihtoehtoa integraatiokuormien käsittelyyn (kts. Timin blogi aiheesta tästä). Nämä no-/low-code -ratkaisut ovat ihan toimivia, joskin sisältävät tiettyjä rajoitteita, esimerkiksi versiohallinnan ja deployment pipelinejen osalta (Pipeline-objektin vienti Deployment Pipelinen läpi ei päivitä dynaamisesti data sourceja työtilasta toiseen, mikä on lievästi sanottuna hieman raskasta). Pipelinet ja Dataflowt ovat kuitenkin helppokäyttöisä ratkaisuja, ja varsinkin Pipelinellä saa todella näppärästi erilaisia operaatioita siistiin ja helposti ymmärrettävään dataputkeen.

Haastajaksi nousee kuitenkin täysin Notebookeihin perustuva orkestrointi. Tässä päästään hyödyntämään Notebook Utilities -pakettia eli NotebookUtilssia (vanha MSSparkUtils), jolla voidaan siis peukaloida tiedostoja, hakea avaimia Key Vaulteista, ja mikä tärkeintä, ketjuttaa eri Notebookien ajoja sekä perättäin että rinnan. Katso lisää Notebookeista Tommin kirjoittamasta blogista tästä.

Käyn tässä tekstissä läpi hypoteettisen esimerkin siitä, miltä monivaiheinen dataputki näyttäisi sekä Pipelinen että Notebookien kautta ajettuna. Ja tietenkin, että millaisista suorituspaineista erityisesti Pipeline kärsii tosipaikan tullen.

Pipeline

Katsotaan ensin hieman Pipelineä, koska se lienee monelle ensimmäinen askel ETL-prosessin orkestrointiin Fabricin maailmassa.

Esimerkissämme kuvataan yksinkertainen DAG (eli Directed Acyclic Graph) jossa on kolme osaa, NB_1, NB_2 ja NB_3.

• Ensimmäinen luo 10 miljoonaa riviä ja 15 saraketta satunnaista dataa ja kirjoittaa sen Delta-tauluun Lakehouseen.
• Toinen luo taas kerran satunnaisia arvoja ja palauttaa niitä notebookutils.notebook.exit(”exit value”) – komennolla. Tämä on muista irrallinen operaatio.
• Kolmas lukee ensimmäisen kirjoittaman Delta-taulun ja laskee tilastollisia tunnuslukuja, mitkä edelleen palautetaan suorituksen päätteeksi. Kolmas luonnollisesti ajetaan vasta kun ensimmäinen Notebook on valmis.

Simppeli kolmivaiheinen Pipeline

Ajetaan Pipeline, odotellaan hetki, katsotaan kuinka kävi:

Ensimmäinen ajo suoritettu

Vähän vajaa 5 minuuttia koko ”ETL-putken” ajamiseen. Ihan hyvä jos ei ole kiire.

Notebook

Jos joko performanssi tai no-/low-code -ympäristö ei miellytä, on Fabricissa tähän onneksi ratkaisu! Notebookeilla voimme tehdä käytännössä saman orkestraation PySparkilla luomalla DAGin dictionary-tietorakenteena, ja sitten ajelemalla Notebookeja NotebookUtils-kirjastolla. Katsotaan miten se toimisi:

Sama putki Notebookissa

Notebookit NB_1, NB_2 ja NB_3 sijaitsevat samassa työtilassa ja kansiossa. Myös eri työtilassa olevia Notebookeja voidaan ajaa, kunhan määritetään työtila tyylillä:

mssparkutils.notebook.run("Sample1", 90, {"input": 20 }, "fe0a6e2a-a909-4aa3-a698-0a651de790aa") # lopussa työtilan ID

Kun muut Notebookit ovat kunnossa, ajetaan solu ja saadaan seuraavanlainen hieno kuva, exit-valuena saadut tulokset ja suoritusaika (hox: %%time – taikasana solun alussa!):

Simppeli putki Notebookissa

Eli vajaat pari minuuttia samojen Notebookien ajamiseen.

Entä sitten?

Tämä simppeli esimerkki oli tietysti kovin yksinkertainen, mutta mitä jos putkessa tehdään vähän enemmän asioita? Kokeillaan seuraavaa hyvin asiaa havainnollistavaa joskin toki hieman epärealistista skenaariota:

Kuvataan yllä jo esiteltyä kolmivaiheista prosessia sanalla ”DAG”, ja pyöräytetään se pariin kertaan siten että kokeneempikin datainssi menisi todellisessa tilanteessa siitä sekaisin:

• Ajetaan DAG kerran
• Ajetaan NB_1 uudestaan
• Ajetaan NB_1 uudestaan
• Ajetaan DAG
• Ajetaan NB_1
• Ajetaan NB_3

Laskelmiemme mukaan tässä ajetaan siis Notebook 10 kertaa. Pipelinessä prosessi rakentuu näin:

Pidempi putki Pipelinessä

Notebookissa määritelmä näyttää tältä:

%%time

# Define initial parameters
p1 = 10000000 # rows to table
p2 = 1337 # random seed for np

BIG_DAG = {
  "activities": [
    {
      "name": "Initial_NB_1",
      "path": "NB_1_final",
      "timeoutPerCellInSeconds": 120,
      "args": {"p1": p1, "p2": p2}
    },
    {
      "name": "Initial_NB_2",
      "path": "NB_2_final",
      "timeoutPerCellInSeconds": 120,
      "args": {"p1": p1, "p2": p2},
      "retry": 1,
      "retryIntervalInSeconds": 10
    },
    {
      "name": "Initial_NB_3",
      "path": "NB_3_final",
      "timeoutPerCellInSeconds": 120,
      "dependencies": ["Initial_NB_1"],
      "args": {"p1": p1, "p2": p2},
      "retry": 1,
      "retryIntervalInSeconds": 10
    },
    # Then run NB_1 twice
    {
      "name": "NB_1_second",
      "path": "NB_1_final",
      "timeoutPerCellInSeconds": 120,
      "dependencies": ["Initial_NB_3"],
      "args": {"p1": p1, "p2": p2}
    },
    {
      "name": "NB_1_third",
      "path": "NB_1_final",
      "timeoutPerCellInSeconds": 120,
      "dependencies": ["NB_1_second"],
      "args": {"p1": p1, "p2": p2}
    },
    # Run the initial DAG again as a sub-process:
    {
      "name": "Second_DAG_NB_1",
      "path": "NB_1_final",
      "timeoutPerCellInSeconds": 120,
      "dependencies": ["NB_1_third"],
      "args": {"p1": p1, "p2": p2}
    },
    {
      "name": "Second_DAG_NB_2",
      "path": "NB_2_final",
      "timeoutPerCellInSeconds": 120,
      "dependencies": ["NB_1_third"],
      "args": {"p1": p1, "p2": p2},
      "retry": 1,
      "retryIntervalInSeconds": 10
    },
    {
      "name": "Second_DAG_NB_3",
      "path": "NB_3_final",
      "timeoutPerCellInSeconds": 120,
      "dependencies": ["Second_DAG_NB_1"],
      "args": {"p1": p1, "p2": p2},
      "retry": 1,
      "retryIntervalInSeconds": 10
    },
    # Finally, run NB_1 followed by NB_3
    {
      "name": "NB_1_fourth",
      "path": "NB_1_final",
      "timeoutPerCellInSeconds": 120,
      "dependencies": ["Second_DAG_NB_3"],
      "args": {"p1": p1, "p2": p2}
    },
    {
      "name": "Final_NB_3",
      "path": "NB_3_final",
      "timeoutPerCellInSeconds": 120,
      "dependencies": ["NB_1_fourth"],
      "args": {"p1": p1, "p2": p2},
      "retry": 1,
      "retryIntervalInSeconds": 10
    }
  ],
  "timeoutInSeconds": 43200,
  "concurrency": 3
}

results_BIG = mssparkutils.notebook.runMultiple(BIG_DAG, {"displayDAGViaGraphviz": True})

print("\nExecution complete. Results:")
for notebook, status in results_BIG.items():
print(f"Notebook {notebook}: {status}")

Ja vielä NotebookUtilssin piirtämä kuva:

Graphvizin upea DAG-kuva

Tulokset

Pipelinessä koko viritelmä vei arvokasta aikaamme 15 minuuttia ja 57 sekuntia.

Notebook suoriutui vastaavasta ajassa 5 minuuttia ja 56 sekuntia.

Pidempi Pipeline-prosessi

Pidemmän prosessin suoritusajat Notebookissa

Käytetyn laskentatehon eli Capacity Unitien (CU) osalta eroa ei juurikaan ollut. Tulkitaan siis asia niin, että Notebook osaa käskyttää laskentaresursseja tehokkaammin, ja täten suoriutuu tehtävästä nopeammin. Lopullinen lasku on kuitenkin maksajalle sama, eli Notebookeilla saamme siis saman prosessin tehtyä samaan hintaan, mutta vauhdikkaammin.

Neljä alinta riviä on Pipeline-ajoja, ylin rivi “master_long_dag_final” Notebook-ajo. Yhteenlaskettu CU about sama, Duration huomattavasti eri

Kumpi voittaa?

Jokainen tehköön omat päätelmänsä tästä, mutta kirjoittajan näkemys on seuraava: jos prosessi on vähänkään raskaampi ja nopeus on kiinnostava tekijä, suositaan Notebookeihin perustuvaa ETL-putkea. Bonuksena tietenkin myös saumaton git-integraatio ja selkeä, helposti monistettava parametrisointi.

Jos arvostetaan käyttöliittymän selkeyttä ja no-/low-code-devauskokemusta, kyllä Pipeline on hyvä. Esimerkiksi Copy data-aktiviteetti on oikeasti näppärä, ja Semantic model refreshille (preview!) ei tietääkseni löydy korvaajaa minkä voi ajaa Notebookin sisällä.

On siis asioita mitä ei joko voi tai ehkä kannata tehdä NotebookUtilsseilla, mutta aika ison osan ETL:stä voi. Kirjoituksen pointtina ei siis suinkaan ole määrätä miten asiat tehdään ”oikein”, vaan havainnollistaa mikä ero suorituskyvyssä on Pipelinen ja Notebookien välillä.

Aurinkoista kevättä kaikille!

P.S. Mitä jos haluamme ajaa useamman DAGin rinnakkain, ei peräkkäin? Tähän löytyy vastaus concurrent.futures -paketin ThreadPoolExecutorilta. Tästä lisää myöhempänä ajankohtana!