Sharding a’la Virgo

A feladat, nagyon nagy mennyiségű adat tárolása – esetünkben 30G rekord –, viszont az adatok rendelkeznek azzal a kellemes tulajdonsággal, hogy bizonyos szempont szerint szeparálhatók.

Ilyen esetekben jöhet szóba valamiféle sharding megoldás, más néven horizontális particionálás. Az ilyen megoldások lényege, hogy az adatokat fizikailag külön adatbázisban tároljuk. A szükséges adatokat az egyes shardokról külön-külön visszanyerjük, majd elvégezzük az esetleges utófeldolgozást (pl.: a különböző shard-okról származó adatok rendezése). Itt rögtön meg is mutatkozik, hogy mikor nem jó a sharding technológia: ha az utófeldolgozás túl költséges.

A sharding-ot legtöbbször alkalmazás szinten valósítjuk meg, bár vannak alacsonyabb szintű kezdeményezések is, például a MySQL Proxy-n alapuló HScale. Ezektől most tekintsünk el, és nézzük, milyen lehetőségeink vannak alkalmazás oldalon. A környezet, amiben dolgozunk: Java 1.5, JPA, Hibernate, MySQL. Hibernate-hez van egy „dobozolt” sharding megoldás, a Hibernate Shards. Ez még béta állapotban van, de nagyon ígéretes. Azért érdemes elolvasni a „Limitations” fejezetet, és figyelni arra, hogy csak akkor lehet eredményesen használni, ha már a tervezésnél figyelembe vesszük, hogy shard-olt adatbázissal fogunk dolgozni. A Shards ugyan rengeteg dolgot megold utófeldolgozással, de ez legtöbbször performancia okokból nem vállalható. Az egyetlen járható út, ha már a tervezés során úgy építjük fel az alkalmazásunkat, hogy elkerüljük a költséges utófeldolgozások szükségességét!

Amiért nem a Shards-ot választottuk, annak két fő oka volt:

• Shard-onként külön SessionFactory-t használ, ami elég nagy erőforrást emészt fel, ráadásul kényelmetlenné teszi a konfigurálást. (A Google saját bevallása szerint körülbelül 10 shard-ig eredményes.)
• Mi JPA-n keresztül szeretnénk használni a Hibernate-et, míg a Shards leginkább a Criteria API-ra van kihegyezve

A mi ötletünk lényege az volt, hogy a SessionFactory-nál mélyebb szinten, a ConnectionProvider interfészen keresztül avatkozunk be. A Hibernate ezen az interfészen keresztül kér el a JDBC connection-öket. Ha itt mindig a megfelelő shard-hoz tartozó connection-t adunk vissza, akkor az egyes műveletek a megfelelő adatbázisszerveren hajtódnak végre. Erőforrásban így shard-onként csak a ConnectionPool-ok többszöröződnek, ami nem kis költség, de nehezen megúszható. Másrészt a sharding miatt egy pool-ban kevesebb connection-lesz a szokásosnál, így kevesebb memóriát emészt fel, és a karbantartó szálakon is spórolhatunk.

A konfigolhatóság egyszerű lesz, hisz a Hibernate által használt ConnectionProvider-t egyszerűen megadhatjuk a szabványos hibernate.connection.provider_class tulajdonságon keresztül (ahogy a ConnectionPool-t is), az egyes shard-okhoz tartozó beállításoknál pedig megtartjuk a Hibernate-es kulcsokat, azzal a megkötéssel, hogy mindegyik elé egy virgo.shard.<SHARD-NEVE>. prefix kerül, és ezt a saját Provider-ünkben fordítjuk vissza a hibernate-nek megfelelő formába. Ezt meg lehet még spékelni default érték kezeléssel (virgo.shard.DEFAULT. prefixű paraméterek minden shard-ra igazak, ha ott nincsenek megadva), hisz shard-onként rengeteg közös paraméter lesz – általában csak a connection url más. Így shard-onként egy plusz konfig sorral elintézhető a beállítás.

Két problémát kellett megoldanunk:

1. Az alkalmazás csak olyan műveleteket hajthat végre, amik csak egyetlen shard-ra vonatkoznak. Minden más műveletet felsőbb rétegben szét kell bontani shard-onkénti műveletekre.
2. A felsőbb szinten valahogy jelezni kell, hogy melyik Shard-on is kell végrehajtani az adott műveletet.

Az 1. pont igen nagy megkötésnek tűnik, azonban éles helyzetben nem igen képzelhető el, hogy olyan lekérdezések futnak, amik több – esetleg minden – shard-ot érintenek, ekkor ugyanis nem sokat nyerünk a sharding-gal. Ebben az esetben valószínű érdemesebb más terheléselosztást választani, esetleg más szisztéma szerint shard-olni. Ha néhány ilyen helyzet mégis előfordul, akkor könnyű rá valamiféle JPA kompatibilis, újrafelhasználható megoldást írni, például QueryBuilder-ek / speciális Shard-olt Query implementációk formájában. Általános megoldás egyébként sincs, ugyanis alacsony szinten egy összetettebb lekérdezésnél igen nehéz eldönteni, hogy mit is akart a hívó. (A Hibernate Shards is korlátozott képességekkel rendelkezik ezen a téren.)

A mi esetünkben nem is volt olyan olvasó művelet, amikor több shard-hoz kellett nyúlni. Író műveletünk volt, ott egy bulk insert-et kiadó QueryBuilder-t valósítottunk meg, ami előbb memóriában összegyűjti a beszúrandó adatokat shard-onként, majd egyesével elküldi őket. Itt problémás a hibakezelés: mi van, ha az n-edik shard tranzakciója nem sikerül? Mi amellett döntöttünk, hogy a query objektum visszaadja a sikertelen rekordokat, ezek egy hibasorra kerülnek, és később újra megpróbáljuk a feldolgozást.

A 2. problémára adott megoldások igazából egy dologra épülhetnek,mivel a ConnectionProvider getConnection() metódusa nem kap paramétereket, így a hívóval csak ThreadLocal-on keresztül kommunikálhat. Arra persze már több lehetőség kínálkozik, hogy ezt a ThreadLocal-t hogyan állítjuk be. Mi egy JPA metodikába illő megoldást kerestünk, a következőt választottuk. Az EntityManager interfészt kiterjesztettük ShardedEntityManager néhány shard-kezelő metódussal, például:

Látható, hogy az egyes entity-khez tartozó shard kiválasztását (resolveShard) és az aktuális shard beállítását (setCurrentShard) külön metódusba tettük. Így lehetséges, hogy eltároljuk a shard nevét, megspórolva a resolve költségét, illetve van lehetőség iterálni az összes shard-on. Annak a leprogramozása, hogy az egyes Entity-khez milyen shard-ok tartoznak egy külön osztályba került, ezt PersistenceUnit-onként egyszer kell megtenni.

Egy statikus metóduson keresztül ­– ShardManager.extend(EntityManager em) –tetszőleges EntityManager kiterjeszthető ShardedEntityManager-ré, ezért került bevezetésre az isSharded() metódus, ami nem shard-olt esetben false-t ad vissza.

A shard-olt eset az érdekesebb, ilyenkor ugyanis nem elég, hogy a visszaadott manager a setCurrentShard metódusban átállítja a ThreadLocal értékét, hiszen ha egy szál esetleg több EntityManager-t használ felváltva, akkor függetlenül a sorrendtől az egyes manager-eknek a rajtuk beállított shard-hoz kell menniük. Ezért egy olyan EntityManager kerül visszaadásra, ami wrap-eli a kapott EntityManager-t, és az eredeti metódusok hívása előtt mindig beállítja az adott manager aktuális shard-ját a ThreadLocal-ban. Mi ezt a wrap-elést CGLib-en keresztüli proxy-zással valósítottuk meg, de meg lehet írni a wrapper osztályt kézzel is.

Megjegyzések:

1. A shard-ok szerinti connection visszaadás nem okoz gondot a 2nd level cache-ben, hisz a shard-ok diszjunktak, az id-k globálisan egyedik.
2. Nem szóltunk az id generálás problémáiról, mivel ez nem tartozik szorosan a témához, de garantálni kell, hogy globálisan egyedi id-k kerülnek kiosztásra. Erre több lehetőség kínálkozik:
   • globális szekvenciából / táblából – pool-ozva – osztjuk az id-t
   • lokálisan osztjuk, de úgy, hogy globálisan egyedi legyen (pl.: <shard-index> + N * <shardok-száma>, ahol N=1, 2, …)
   • összetett kulcs (shard + lokális id)
3. A kézi shard állítás automatizálásán is gondolkodtunk, több ötletünk is volt, de egy olyan sem, ami elég általános, és valódi haszonnal járó lett volna. Ami nekünk eszünkbe jutott:
   • ha shard-olt entity-vel hívunk persist(), merge(), stb. metódust, akkor a shard automatikusan állítódjon be.
   • findEntity(@Shard(class=Entity.class) int primaryKey)
   • A fenti üzleti metódus esetén az aktuális Shard beállítódik a class, és primaryKey által meghatározott entity-nek megfelelőre.
4. Természetesen a query-k „shard-osításán” is gondolkodtunk, de ennek általános megoldása túl nagy falat – nem is volt cél –, és performancia szempontból sem tűnik jó útnak. Egy egyszerű példa: már egy query Order By feltételének megfelelő Comparator elkészítése, amivel ugyanaz a rendezés a memóriában elvégezhető sem egyszerű. (Nem is beszélve arról, ha az oszlop, ami szerint rendezünk, nem is szerepel a visszaadott mező listában; a lokalizációs problémákról, stb.)