Pyspark kalandok

A data team megörökölt egy masszív adatfeldolgozó pipeline-t, amit egy genetikus csávó írt. Nagyon tehetséges a fickó, de az a baj, hogy az egész egy prototípus: a pipeline fragmentált, az egyes komponensek nem robosztusak, nem optimalizált se a kód, és úgy általában maga az adat flow sem.

Eddig nem volt túl nagy gond, leszámítva, hogy másfél évig nem volt ember aki futtassa, miután a gyerek lelépett tőlünk. :D Most ott tartunk, hogy az utóbbi release nagyjából 3 hónap alatt jött össze. Ez már mindenkinél kiverte a biztosítékot, úgyhogy megszületett az executive döntés: a data team átveszi a genetics team-től az egész hóbelevancot. A cél az, hogy production szintre hozzuk: legyen a kódbázis konzisztens, optimalizált, könnyen bővíthető, és a lehető legkisebb emberi beavatkozással, a lehető legrövidebb idő alatt legyen release.

Első lépésként az egész release leglassabb, colokalizációnak nevezett folyamatát vettük szemügyre. A folyamat optimalizálása azért kritikus, mert itt az adatmérettel négyzetesen nő a feldolgozási igény: minden genetikai vizsgálat, minden eredményét hasonlítjuk össze minden más vizsgálat minden eredményével. Az aktuális implementáció szerint file-okat írunk diszkre, amikre meghívunk egy C-ben írt alkalmazást, ami elvégzi a tesztet, aminek az eredményét szintén diszkre írja, amit be kell olvasni. Ez egy teljesen normális eljárás, ha az embernek van mondjuk 5 vizsgálata amiket egyenként command line-on futtat le. A script eredeti fejlesztője ezt a logikát követte. A mostani adatszett viszont már milliós méretű, a potenciális tesztek száma csillagászati. Egy kétszáz core-os gépen több, mint egy napig ketyeg.

A megoldást végül az hozta, hogy az új genetikusunk talált egy új, Cambridge-i egyetemen fejlesztett módszert, ahol az algoritmus R-ben implementálva megtalálható github-on és kellően egyszerűnek tűnt, hogy átírjuk pySpark-ra. Rengeteget tanultam ebből az erőfeszítésből. Ugyan rutinszerűen használunk pySpark-ot főleg ETL szerű dolgokra: joinok, aggregációk, stb. Itt viszont egy relatíve komplikált algebrai műveleteket kellett megvalósítani. Első nekifutásra kb. egy teljes napot azzal töltöttünk, hogy egy tábla előtt nyilazgattuk, hogy hol milyen táblázatot hogyan join-oljuk, és aggregáljunk, hogy a megfelelő számok legyenek egymás mellett. Ezzel tudtuk kihasználni a Spark igazi erősségét, a jól párhuzamosítható oszlopműveleteket.

Az eredmény az lett, hogy egy 16 core-os gépen lefutott 20 perc alatt. Mindenkinek leesett az álla.... sajnos aztán kiszúrtam, hogy a kelleténél több kanyart vágtunk le, és olyan egyszerűsítéseket is megengedtünk magunknak, amire nem volt okunk. Újabb kör reszelés, aminek a vége az lett, hogy a 16 core-os gépen lefut 13 óra alatt. Ugyan most az eredmények teljesen egybevágtak az R script által adott számokkal, de nem volt elég gyors... gondolkoztunk rajta sokat, és végül arra jutottunk, hogy az a baj, hogy algebrai műveleteket spark array objektumokon hajtunk végre, ami nem optimális. Ehelyett a spark.ml dense vektorokkal kell dolgozni, és a számításokat ugyancsak spark.ml függvényekkel kell végezni. Szerintem emiatt kicsit csúnyább így a kód, de végülis 4 órás futásidőt kaptunk. Öröm és boldogság.

 

Tanulságok:

  • Ha tényleg komolyan optimalizálni kell, akkor megéri sok időt arra fordítani, hogy az ember gondolkodjon a megfelelő adatmodellen. Az intuíció nem feltétlenül működik, ha több kell, mint egy "good enough" megoldás. (De lehet, hogy ebben még nincs elég rutinunk.)
  • Ha optimalizálni kell, akkor az másodlagos, hogy maga az algoritmus mennyire érhető. Ez most kicsit hülyén hangzik... nagyjából azt értem ezalatt, hogy míg az R-ben implementált statisztikai teszt teljesen érthető és követhető, hogy a függvény mit csinál, valahogy a spark implementációban ez teljesen obfuszkált. Szép, olvasható a kód, de, hogy mi miért történik, az teljesen érthetetlen. (pl. egy kiindulási adatszettel dolgozunk, de a script-ben van vagy fél tucat join.)
  • Optimalizációra ad lehetőséget ha megfelelő adattárolót használja az ember: pl az intuitív array helyett az általunk sosem használt dense vector.

 

És ez még csak egy lépés a teljes pipeline-ból... :D 

( sinexton | 2022. 04. 10., v – 08:53 )

Amikor N elemet N elemmel kell hasonlítani, akkor bizonyos esetekben használható, hogy véletlenszerűen mintavételezed az adatot, és a mintára fut az elemezés. A nagy számok törvényének matematikai tétele miatt hasonló tulajdonságokat fog hordozni. Így az N lépés N/M-re csökkenthető, mellyel négyzetesen is visszahúzódik az idő.

Hogy mekkora mintavétel kell, az meg egy power analízissel elvégezhető, ha nem túl sok a változók száma. Vagy akár mindegyik változóra külön elvégezhető, és ezek közül a maximum minta szám is választható.

További lehetőség így távolról ránézve nulla infóból, például MCMC mintavételezés (Markov Chain Monte Carlo). Lényeg hogy szignifikánsnak kell lennie a mintavételezéshez, és ha ezt biztosítjátok, akkor nem kell az összes kombináció. Hogy mekkora megbízhatóságot kaptok M elemmel, az számítható lehet.

Ha töredékére húzódott az idő, akkor meg többször futtatható más mintavételezéssel, mellyel nézhetitek hogy konvergál-e valami felé az eredmény. Ez megérheti és  (N/M)^2 * K gyorsabb lehet mint N^2.

Az általam írt abban az esetben nem jó, ha túl sok az egyedi tulajdonság az adatban.

Szia! Köszi a megjegyzést. Szerencsére van arra lehetőség, hogy egzakt módon leszűkítsük az elméletileg nxn méretű teret: a genetikai adatok, amivel dolgozunk emberi GWAS associációt, szóval elég csak az átfedő szignálokat vizsgálni. Viszont ahol van átfedés, azt ellenőrizni kell. Ezt a célt szolgálja a sok join. A posztban nem linkeltem kódot, mert egyelőre egy gist formájában létezik, de ha van kedved belenézhetsz.

Ugyan ez elvégzendő tesztek száma nem nxn, de a tesztek számának növekedése az adatszett méretével így is négyzetesen nő. Felmerült, hogy esetleg inkrementálisan kellene a számolásokat elvégezni: mindig csak azt kiszámolni, hogy az előző release óta megjelenő adatok hol kolokalizának a már meglévő adatokkal. Ez nagyon gyors lett volna, de a logisztika borzalmasan komplikáltá és törékennyé vált, így ezt elvetettük.

( sinexton | 2022. 04. 10., v – 13:47 )

Még egy ötlet: ha van rá lehetőség, akkor majdnem teljesen linearizálható az összehasonlítás úgy, hogy kevert sorrenden az egymás mellett lévőket hasonlítjátok össze. A véletlen sorrend garantálja a varianciát.

( Nyosigomboc v | 2022. 04. 10., v – 17:44 )

Mindenkeppen erdekes a tavoli elemek kozti tavolsag is?

Ja van 3 elemed: egy sargarepa, egy nyul, meg egy E. Coli, akkor ha bejon egy masik nyul, eleg lehet a mar meglevo nyulhoz hasonlitani, a sargarepa meg az E. Coli mar elegge tavoli. Ezt a besorolast meg klaszterezessel gyorsan meg lehet tenni, es csak a kozeli elemekkel/klaszterekkel kell csak foglalkozni az N**2 algoritmus helyett.

A strange game. The only winning move is not to play. How about a nice game of chess?

Ezek kizárólag humán, úgynevezett GWAS (genome-wide association studies) vizsgálatokból származó adatok. Tulajdonképpen azt nézzük, hogy pl. egy elhízást kutató vizsgálat által felfedezett asszociációk közül melyek fednek át a hasnyálmirigy rák vizsgálatban felfedezett asszociációkkal. De itt is ugye ki lehet zárni, ha a felfedezett helyek a genomban túl távol állnak egymástól. Viszont ahol van átfedés, ott meg kell vizsgálni, hogy az átfedés azt jelzi-e, hogy a két szignálnak azonos a genetikai háttere.