Agregace velkého množství streamovaných dat

[Standa Blábol]

Ale jako prvni pokus zkusim ten MapDB projekt.

nejhorsi mozna volba, kdyz uz chcete ukladat json, tak je efektivnejsi Mongo.

Ok, na ten se taky podivam.
Popravde me efektivita ukladani moc nezajima, chci do toho nacpat hodinu metric dat, pak to poslu do sveta, pak to cely smazu a dalsi hodinu dat.

[Reklama]

[Death Walker]

Ked uz tam mas niekde postgres tak ta asi bude zaujimat toto: https://www.timescale.com/

Robi to presne to co pozadujes. Naviac ti to poskytne efektivnejsie a bezpecnejsie ulozisko ako kafka...

timescale zrovna moc efektivni ve vyuziti mista na disku neni, ma spatnou kompresi, pokud nepotrebujete delat nejake fancy dotazy, tak bych se mu vyhnul.

a nemusi stihat zapis, 1 mld za hodinu je celkem dost dat.

tady jsou nejake benchmarky

https://altinity.com/blog/clickhouse-for-time-series

https://github.com/timescale/tsbs

rozdil celkem vyrazny 26GB timescale vs 0.5GB influx vs 1.2GB clickhouse

neverte vsetkemu co citate na internete... Podla tohoto vychadza Timescale proti Influx vyrazne lepsie: https://blog.timescale.com/blog/how-to-benchmark-iot-time-series-workloads-in-a-production-environment/

Co sa tyka velkosti. Postgres tam uklada aj data, nie len medzivysledky. Naviac by bolo zaujimave kolko by ostalo z tych 26gb po vacuum, pripadne pri zapnutom autovacuum.

Co sa tyka "fancy" dotazov, tak je fajn ze funkcie mozete pisat aj v R, pripadne v pythone a vyuzit numpy. Priklad: https://docs.timescale.com/timescaledb/latest/tutorials/time-series-forecast/#seasonal-arima-with-r

[Google CTCCTCGGCGGGCACGTAG]

Ked uz tam mas niekde postgres tak ta asi bude zaujimat toto: https://www.timescale.com/

Robi to presne to co pozadujes. Naviac ti to poskytne efektivnejsie a bezpecnejsie ulozisko ako kafka...

timescale zrovna moc efektivni ve vyuziti mista na disku neni, ma spatnou kompresi, pokud nepotrebujete delat nejake fancy dotazy, tak bych se mu vyhnul.

a nemusi stihat zapis, 1 mld za hodinu je celkem dost dat.

tady jsou nejake benchmarky

https://altinity.com/blog/clickhouse-for-time-series

https://github.com/timescale/tsbs

rozdil celkem vyrazny 26GB timescale vs 0.5GB influx vs 1.2GB clickhouse

neverte vsetkemu co citate na internete... Podla tohoto vychadza Timescale proti Influx vyrazne lepsie: https://blog.timescale.com/blog/how-to-benchmark-iot-time-series-workloads-in-a-production-environment/

Co sa tyka velkosti. Postgres tam uklada aj data, nie len medzivysledky. Naviac by bolo zaujimave kolko by ostalo z tych 26gb po vacuum, pripadne pri zapnutom autovacuum.

Co sa tyka "fancy" dotazov, tak je fajn ze funkcie mozete pisat aj v R, pripadne v pythone a vyuzit numpy. Priklad: https://docs.timescale.com/timescaledb/latest/tutorials/time-series-forecast/#seasonal-arima-with-r

ten vami odkazovany clanek je z blogu timescale, srovnava pouze s influx, nesrovnavaji spotrebu mista na disku, metrika ve ktere timescale prohrava na cele care. 

[Death Walker]

Ale jako prvni pokus zkusim ten MapDB projekt.

nejhorsi mozna volba, kdyz uz chcete ukladat json, tak je efektivnejsi Mongo.

Ok, na ten se taky podivam.
Popravde me efektivita ukladani moc nezajima, chci do toho nacpat hodinu metric dat, pak to poslu do sveta, pak to cely smazu a dalsi hodinu dat.

Tak pozrite este na https://github.com/rrd4j/rrd4j

je to v jave, ak by nestacila pamat tak si ako backend mozete dat subor na disku, mongodb a funguje to ako klasicke rrdtool.

[Death Walker]

ten vami odkazovany clanek je z blogu timescale, srovnava pouze s influx, nesrovnavaji spotrebu mista na disku, metrika ve ktere timescale prohrava na cele care.

https://severalnines.com/database-blog/which-time-series-database-better-timescaledb-vs-influxdb chcete dalsie ale si vygooglite sam?

A ak si to neviete najst tak:

• For workloads with very low cardinality (e.g., 100 devices), InfluxDB outperforms TimescaleDB.
• As cardinality increases, InfluxDB insert performance drops off faster than on TimescaleDB.
• For workloads with moderate to high cardinality (e.g., 100 devices sending 10 metrics), TimescaleDB outperforms InfluxDB.
  

Ad miesto na disku, timescaledb je OLAP pre postgres, podporuje transakcie, referencnu integritu a podobne. Influx je nosql, len tak volne pohodena halda dat.

[Reklama]

[Idris]

Nez C++, to bych to pro tento ucel radsi delal v GO.

Go bude mít sice menší memory footprint, ale v případě spousty malých objektů na haldě člověk pak stejně musí dělat stejné opičárny jako v Javě. Pokud přes C++ nejede vlak, tak se v tomto případě nabízí Rust.

[sofa_king]

Use case je pullnout každou hodinu data z kafky, agregovat je, zapsat jinam a zdroj zahodit. A tak dokola.

Pattern zní mini-batching. Ie. batch job jednou za hodinu.

Pokud máš metrik jenom pár, bez žádného group by, streamneš to jedním loopem v několika paralelních procesech třeba v pythonu nebo javě a metriky za běhu sečteš pomocí hashmapy a na konci vyplivneš.

V tvým případě není problém spousta metrik (counters), ale počet dimenzí (metadata), kvůli kterým ti vzroste náročnost na paměť pro hashmapu, do kterých ukládáš mezivýsledky. Kardinalita dimenzí a metrik ti přímo určí, kolik paměti potřebuješ. Pronásob si kardinalitu všech dimenzí a metrik a když se ti to vejde do paměti, tak to akorát loopni skrz hashmapu a hotovo.

Pokud ne, dumpni všechny hodnoty za hodinu do csv souboru, seřaď vše podle dimenzí pomocí externího sort algoritmu (řazení mimo RAM), např. pomocí standardního GNU sort. Pak to loopni v javě podobně jako v prvním případě, ale vyhazuj položky z hashmapy tak, jak je navštěvuješ.

[honzanovak555]

Spravne riesenie je pouzit Kafka Spark streaming, je to presne robene na to co tazatel potrebuje.

Zakladna idea je ze sa Spark pripoji na Kafku ako consumer a pravidelne taha z Kafky data do Dstreamu nad RDD a nad tym si spravis agregacie absolutne jednoducho. Ono je to uplne usite namieru tomuto problemu.

Tym padom nemusis vobec drzat nikde tie data, staci to on-the-fly agregovat a pamatat si posledny event.

https://spark.apache.org/docs/latest/streaming-kafka-0-10-integration.html

https://spark.apache.org/docs/latest/streaming-programming-guide.html

[_Tomáš_]

a Kafka Streams a inkrementální agregace ti nestačí?

Základem je co nejvíce metrik převést na inkrementální (stejný problém řeší např. GoodData), poté počítání v paměti je snadné. Ne vše držet v paměti jednoho procesu, lepší je využít nějakou databázi, např. redis v tomhle poslouží dobře.

Tenhle problém v enterprise často řešíme buď různými obskurně dranými Oracle/SAP nástroji nebo to kluci programují přes spark streams.

Clickhouse není vhodný na příliš mnoho insertů a potřebuje data v bulku, jinak moc dobře s tím nepracuje.

[M_D]

Také bych viděl toto jako "typickou" úlohu pro Apache Spark.

AoK: Redis bych do tohoto netahal. To už možná spíše pak Apache Ignite - umí se sám napojit na Kafku a na příchozí data, co strká do sebe, volat transformační funkce (případně i ten Spark). A na druhé straně to umí i plivat ven dále. A pokud by těch metrik bylo opravud hodně, tak Ignite umí udělat rozumnou distribuci mezi víc serverů nebo ho místo RAM only DB můžu použít s disk backendem, což bych si u Redisu musel řešit svépomocí.

[honzanovak555]

Doplnim ze Ignite ma IgniteRDD na ktore je mozne zavesit Spark a spracovat to ale naopak to nefuguje - Spark nevie spravit z Ignite sink.

Kafka ako source pre Ignite a tam to spracovat je urcite zaujimave ale pride mi to ako kanon na vrabce. Zaroven Ignite sa mi zda nema tie time window funkcie take pohodlne ako Spark.

Ja som pouzival raz Ignite ako write through cache pre Cassandru a potom sa nad Ignitom v cachi da volat normalne (distribuovane) sql s joinami co sa v Cassandre neda.

[František Ryšánek]

Kód: [Vybrat]

{
"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,
...
...Tam muze opravdu byt libovolny bordel a v libovolnem poradi...

Tenhle konkrétní datový formát jsem nepotkal, přece jenom už mnoho let "dělám do včel". Řekl bych, že jde o nějakou telemetrii ze skupiny nějakých "aktivních prvků" (A/V komunikačních zařízení). Jiní zde asi vidí na první pohled, o co přesně jde a znají hotová řešení.

Spíš pro svou zábavu a pro zajímavost bych se rád zeptal na upřesnění "metamodelu" těch dat (a taky protože jsem možná natvrdlej):

Vypadá to na nějaký polymorfní objekt zvaný "metric", kde "metadata" je stálá/společná množina čtyř atributů, ale polymorfní je množina atributů "counters". Mám pravdu? A množina objektů "counters" je stabilní alespoň pro konkrétní "detailní třídu metriky", tzn. pro konkrétní hodnotu klíčového atributu "metric" ? Tzn. pokud byste si vedl mapu klíčovanou polem "metric", tak jednotlivé uzly (záznamy v mapě) už budou mít každý svou stálou množinu atributů "counters"? Nebo i pak jsou přípustné prázdné hodnoty? (pořadí counterů neřeším.)
Různých typů objektu "metric" je cca kolik? Chápu že 90 milionů je jich *kusů* (úhrnem instancí).

Protože držím v ruce kladivo, problém mi připomíná hřebík: v C++ bych na to použil dvě patra indexu s použitím třídy "std::map", možná se substitucí klíčových stringů integerem, pokud bych usoudil, že to přináší nějakou výkonovou výhodu.

[František Ryšánek]

cat input | grep metric | sort | uniq | wc -l

Jo a agregace podle zařízení / karty / portu by přidala další zábavné patro indexu.
Taky by se dala udělat jediná plochá mapa, tříděná podle třeba pěti kritérií ;-)
Nebo ten model nějak setřepat/optimalizovat.

[Ladislav Zitka]

Spark standalone mode, single worker bych zkusil jako poc

[Standa Blábol]

Kód: [Vybrat]

{
"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,
...
...Tam muze opravdu byt libovolny bordel a v libovolnem poradi...

Tenhle konkrétní datový formát jsem nepotkal, přece jenom už mnoho let "dělám do včel". Řekl bych, že jde o nějakou telemetrii ze skupiny nějakých "aktivních prvků" (A/V komunikačních zařízení). Jiní zde asi vidí na první pohled, o co přesně jde a znají hotová řešení.

Spíš pro svou zábavu a pro zajímavost bych se rád zeptal na upřesnění "metamodelu" těch dat (a taky protože jsem možná natvrdlej):

Vypadá to na nějaký polymorfní objekt zvaný "metric", kde "metadata" je stálá/společná množina čtyř atributů, ale polymorfní je množina atributů "counters". Mám pravdu? A množina objektů "counters" je stabilní alespoň pro konkrétní "detailní třídu metriky", tzn. pro konkrétní hodnotu klíčového atributu "metric" ? Tzn. pokud byste si vedl mapu klíčovanou polem "metric", tak jednotlivé uzly (záznamy v mapě) už budou mít každý svou stálou množinu atributů "counters"? Nebo i pak jsou přípustné prázdné hodnoty? (pořadí counterů neřeším.)
Různých typů objektu "metric" je cca kolik? Chápu že 90 milionů je jich *kusů* (úhrnem instancí).

Protože držím v ruce kladivo, problém mi připomíná hřebík: v C++ bych na to použil dvě patra indexu s použitím třídy "std::map", možná se substitucí klíčových stringů integerem, pokud bych usoudil, že to přináší nějakou výkonovou výhodu.

Vicemene ano, jenom tech metadat je v realu mnohem vic. Pro agregaci je nepotrebuju, ale potrebuju znat vsecha metadata z posledni zpravy, co odpovida agregacni kombinaci.

Vysvetleno v SQL, potrebuju select sum,count,min,max from table where timestamp between x,x+1 group by some_metadata. A pak jeste kompletni metadata (ne jenom ty v group by klauzuli) casove posledniho radku pro kazdou group by grupu.

C++ pristup pres Hashmapy vybehne z RAM.