Dotaz do think tanku na moznosti zpracovani velkeho mnozstvi stream dat.
Kafka mi posila stream metric dat v JSON formatu.
Dat je hodne, 90 mil ruznych typu metrik, kazda prijde s periodou 5 minut, tedy za hodinu 90*12=1.08 miliardy zprav.
Format (zjednoduseny) je takovyto, realna jedna zprava ma cca 1.5kB dat.
{
"metric": "metric1"
"metadata": {
"nodeid": "node1",
"cardid": "card1",
"portid": "port1",
"lvidid": "lvid1"
},
"counters": {
"MP_TOTAL_TRANS_AUDIO_SESSIONS": 0,
"MP_PEAK_TRANS_VIDEO_SESSIONS": 0,
"MP_ACTIVE_TRANS_DTMF_SESSION": 0,
"MP_ACTIVE_AUDIO_SESSIONS": 3,
"MP_PEAK_SRTP_SESSIONS": 0,
"MP_PEAK_TRANS_AUDIO_SESSIONS": 0,
"MP_TOTAL_SRTP_SESSIONS": 0,
"MP_PEAK_TRANS_DTMF_SESSION": 0,
"MP_TOTAL_AUDIO_SESSIONS": 18215,
"MP_TOTAL_VIDEO_SESSIONS": 6,
"MP_PEAK_AUDIO_SESSIONS": 14,
"MP_PEAK_VIDEO_SESSIONS": 2,
"MP_ACTIVE_TRANS_AUDIO_SESSIONS": 0,
"MP_ACTIVE_SRTP_SESSIONS": 0,
"MP_TOTAL_ROGUE_SESSIONS": 0,
"MP_ACTIVE_VIDEO_SESSIONS": 0,
"MP_ACTIVE_TRANS_VIDEO_SESSIONS": 0,
"MP_TOTAL_TRANS_VIDEO_SESSIONS": 0
}
}
Mam existujici (muj) Java programek, ktery se napoji na kafku, nacita zpravy, mirne transformuje a posila na soket k dalsimu zpracovani. Load to zvlada bez problemu.
Nyni potrebuju nad temito daty provadet hodinove agregace, pricemz dale potrebuju agregovat podle subsetu metadat.
Pro priklad dat vyse, potrebuju agregovat countery (min/max/avg) pro kombinaci metadat nodeid,cardid,portid - hodnota lvidid se muze menit, potrebuju zachovat hodnotu lviid z posledni prijate zpravy pro kombinaci nodeid,cardid,portid.
Vysledkem ma byt obdobny JSON, kde blok metadat bude obsahovat metadata z posledni zpracovane zpravy a blok agregovanych counteru bude pro kazdy counter obsahovat sadu agregovanych dat sum/count/min/max - (avg=sum/count). Nejak takhle:
{
"metric": "metric1"
"metadata": {
"nodeid": "node1",
"cardid": "card1",
"portid": "port1",
"lvidid": "lvid23" # hodnota metadat z posledni zpravy
},
"counters": {
"MP_TOTAL_TRANS_AUDIO_SESSIONS": [120, 20, 0, 30], # sum/count/min/max
"MP_PEAK_TRANS_VIDEO_SESSIONS": [130, 25, 0, 32],
.
.
.
}
}
Muj prvni naivni navrh je:
- pustim kafka replay od hodiny X do X+1
- v mym Java programku na cteni kafka streamu si udelam HashMapu, kde key bude kombinace nodeid,cardid,portid a value bude bean se string metadata (raw JSON) a HashMapa <countername, aggvalues array>
- pro kazdou prichozi zpravu si sestavim key (nodeid,cardid,portid) a v HashMape pro dany key vytvorim/updatuju bean (metadata string se preplacne, agregovane hodnoty prepocitaji)
- az probehnu cely kafla replay - vysmahnu ven HashMapu v JSONu
Problem je s velikosti dat. Vyse popsanou agregaci se dostanu na cca 70 milionu keys v HashMape. Pokud pocitam, ze jeden value v HashMape zabere cca 2kB RAM, celkem ta HashMapa zabere 130GB RAM - nesmysl.
Muj druhy naivni navrh, jako HashMapu pouziju Postgresa, ve kterym bude jedna tabulka reprezentujici vyse popsanou HashMapu ve formatu
(
key varchar(255),
aggdata jsonb
)A na postgresu bude pro insertovani PLSQL procedura, ktera provede potrebne vytvoreni/update pole "aggdata jsonb"
Velikost dat by postgres zvladnout mel, jak to bude s rychlosti, netusim.
Prosim, budu vdecen za jakekoliv hinty, jak toto resit, dik.
60 Odpovědí
29664 Zhlédnutí