మీరు KDB+ బేస్, Q ప్రోగ్రామింగ్ లాంగ్వేజ్ అంటే ఏమిటి, వాటి బలాలు మరియు బలహీనతలు ఏమిటి అనే దాని గురించి మీరు చదవగలరు. మరియు పరిచయంలో క్లుప్తంగా. కథనంలో, మేము Qలో సేవను అమలు చేస్తాము, అది ఇన్‌కమింగ్ డేటా స్ట్రీమ్‌ను ప్రాసెస్ చేస్తుంది మరియు "రియల్ టైమ్" మోడ్‌లో ప్రతి నిమిషానికి వివిధ అగ్రిగేషన్ ఫంక్షన్లను గణిస్తుంది (అనగా, డేటా యొక్క తదుపరి భాగానికి ముందు ప్రతిదీ లెక్కించడానికి సమయం ఉంటుంది). Q యొక్క ప్రధాన లక్షణం ఏమిటంటే ఇది వెక్టర్ భాష, ఇది ఒకే వస్తువులతో కాకుండా వాటి శ్రేణులు, శ్రేణుల శ్రేణులు మరియు ఇతర సంక్లిష్ట వస్తువులతో పనిచేయడానికి మిమ్మల్ని అనుమతిస్తుంది. Q మరియు దాని బంధువులు K, J, APL వంటి భాషలు వాటి సంక్షిప్తతకు ప్రసిద్ధి చెందాయి. తరచుగా, జావా వంటి సుపరిచితమైన భాషలో కోడ్ యొక్క అనేక స్క్రీన్‌లను తీసుకునే ప్రోగ్రామ్‌ను వాటిపై కొన్ని పంక్తులలో వ్రాయవచ్చు. ఈ వ్యాసంలో నేను ప్రదర్శించాలనుకుంటున్నాను.
పరిచయం
KDB+ అనేది చాలా పెద్ద మొత్తంలో డేటాపై దృష్టి కేంద్రీకరించబడిన స్తంభాల డేటాబేస్, నిర్దిష్ట మార్గంలో (ప్రధానంగా సమయం ప్రకారం) ఆర్డర్ చేయబడింది. ఇది ప్రధానంగా ఆర్థిక సంస్థలలో ఉపయోగించబడుతుంది - బ్యాంకులు, పెట్టుబడి నిధులు, బీమా కంపెనీలు. Q భాష అనేది KDB+ యొక్క అంతర్గత భాష, ఇది ఈ డేటాతో సమర్థవంతంగా పని చేయడానికి మిమ్మల్ని అనుమతిస్తుంది. Q భావజాలం సంక్షిప్తత మరియు సమర్థత, అయితే స్పష్టత త్యాగం చేయబడింది. వెక్టార్ భాష ఏ సందర్భంలోనైనా అర్థం చేసుకోవడం కష్టమవుతుంది మరియు రికార్డింగ్ యొక్క సంక్షిప్తత మరియు గొప్పతనం ప్రోగ్రామ్‌లోని చాలా పెద్ద భాగాన్ని ఒకే స్క్రీన్‌పై చూడటానికి మిమ్మల్ని అనుమతిస్తుంది, ఇది చివరికి అర్థం చేసుకోవడం సులభం చేస్తుంది.
ఈ కథనంలో మేము Qలో పూర్తి స్థాయి ప్రోగ్రామ్‌ను అమలు చేస్తాము మరియు మీరు దీన్ని ప్రయత్నించవచ్చు. దీన్ని చేయడానికి, మీకు అసలు Q అవసరం. మీరు kx కంపెనీ వెబ్‌సైట్‌లో ఉచిత 32-బిట్ వెర్షన్‌ను డౌన్‌లోడ్ చేసుకోవచ్చు – . అక్కడ, మీకు ఆసక్తి ఉంటే, మీరు Q, పుస్తకంపై సూచన సమాచారాన్ని కనుగొంటారు మరియు ఈ అంశంపై వివిధ కథనాలు.
సమస్య యొక్క ప్రకటన
ప్రతి 25 మిల్లీసెకన్లకు డేటాతో కూడిన పట్టికను పంపే మూలం ఉంది. KDB+ ప్రధానంగా ఫైనాన్స్‌లో ఉపయోగించబడుతుంది కాబట్టి, ఇది లావాదేవీల (ట్రేడ్‌లు) పట్టిక అని మేము ఊహిస్తాము, ఇది క్రింది నిలువు వరుసలను కలిగి ఉంటుంది: సమయం (మిల్లీసెకన్లలో సమయం), sym (స్టాక్ ఎక్స్ఛేంజ్‌లో కంపెనీ హోదా - IBM, AAPL,...), ధర (షేర్లు కొనుగోలు చేయబడిన ధర), పరిమాణం (లావాదేవీ పరిమాణం). 25 మిల్లీసెకన్ల విరామం ఏకపక్షంగా ఉంటుంది, చాలా చిన్నది కాదు మరియు చాలా పొడవుగా ఉండదు. దాని ఉనికి అంటే డేటా ఇప్పటికే బఫర్ చేసిన సేవకు వస్తుంది. ప్రస్తుత లోడ్‌పై ఆధారపడి డైనమిక్ బఫరింగ్‌తో సహా సేవ వైపు బఫరింగ్‌ను అమలు చేయడం సులభం, కానీ సరళత కోసం, మేము స్థిర విరామంపై దృష్టి పెడతాము.
సేవ తప్పనిసరిగా సిమ్ కాలమ్ నుండి వచ్చే ప్రతి నిమిషానికి ఒక సముదాయ ఫంక్షన్‌ల సమితిని తప్పనిసరిగా లెక్కించాలి - గరిష్ట ధర, సగటు ధర, మొత్తం పరిమాణం మొదలైనవి. ఉపయోగపడే సమాచారం. సరళత కోసం, మేము అన్ని ఫంక్షన్లను క్రమంగా లెక్కించవచ్చని ఊహిస్తాము, అనగా. కొత్త విలువను పొందడానికి, రెండు సంఖ్యలను తెలుసుకోవడం సరిపోతుంది - పాత మరియు ఇన్‌కమింగ్ విలువలు. ఉదాహరణకు, ఫంక్షన్లు గరిష్టం, సగటు, మొత్తం ఈ లక్షణం కలిగి ఉంటాయి, కానీ మధ్యస్థ ఫంక్షన్ లేదు.
ఇన్‌కమింగ్ డేటా స్ట్రీమ్ సమయం ఆర్డర్ చేయబడిందని కూడా మేము ఊహిస్తాము. దీంతో చివరి నిమిషంలో మాత్రమే పని చేసే అవకాశం లభిస్తుంది. ఆచరణలో, కొన్ని నవీకరణలు ఆలస్యం అయినప్పుడు ప్రస్తుత మరియు మునుపటి నిమిషాలతో పని చేయగలిగితే సరిపోతుంది. సరళత కోసం, మేము ఈ కేసును పరిగణించము.
అగ్రిగేషన్ విధులు
అవసరమైన అగ్రిగేషన్ ఫంక్షన్లు క్రింద ఇవ్వబడ్డాయి. సేవపై లోడ్ పెంచడానికి నేను వాటిలో వీలైనన్ని ఎక్కువ తీసుకున్నాను:
- అధిక - గరిష్ట ధర - నిమిషానికి గరిష్ట ధర.
- తక్కువ - నిమి ధర - నిమిషానికి కనీస ధర.
- మొదటి ధర - మొదటి ధర - నిమిషానికి మొదటి ధర.
- చివరి ధర - చివరి ధర - నిమిషానికి చివరి ధర.
- మొదటి పరిమాణం - మొదటి పరిమాణం - నిమిషానికి మొదటి వాణిజ్య పరిమాణం.
- lastSize - చివరి పరిమాణం - ఒక నిమిషంలో చివరి వాణిజ్య పరిమాణం.
- numTrades – కౌంట్ i – నిమిషానికి ట్రేడ్‌ల సంఖ్య.
- వాల్యూమ్ - మొత్తం పరిమాణం - నిమిషానికి వాణిజ్య పరిమాణాల మొత్తం.
- pvolume – మొత్తం ధర – నిమిషానికి ధరల మొత్తం, avgPriceకి అవసరం.
- – మొత్తం టర్నోవర్ ధర*పరిమాణం – నిమిషానికి లావాదేవీల మొత్తం పరిమాణం.
- avgPrice – pvolume%numTrades – నిమిషానికి సగటు ధర.
- avgSize – volume%numTrades – నిమిషానికి సగటు వాణిజ్య పరిమాణం.
- vwap - టర్నోవర్% వాల్యూమ్ - లావాదేవీ పరిమాణం ఆధారంగా నిమిషానికి సగటు ధర.
- cumVolume - సమ్ వాల్యూమ్ - మొత్తం సమయంలో లావాదేవీల సంచిత పరిమాణం.
ఈ నిలువు వరుసలను మొదటి సారి మరియు ప్రతి తదుపరి నిమిషానికి ఎలా ప్రారంభించాలో - వెంటనే ఒక స్పష్టమైన విషయం గురించి చర్చిద్దాం. మొదటి ధర రకంలోని కొన్ని నిలువు వరుసలు ప్రతిసారీ శూన్యానికి తప్పనిసరిగా ప్రారంభించబడాలి; వాటి విలువ నిర్వచించబడలేదు. ఇతర వాల్యూమ్ రకాలను ఎల్లప్పుడూ 0కి సెట్ చేయాలి. మిశ్రమ విధానం అవసరమయ్యే నిలువు వరుసలు కూడా ఉన్నాయి - ఉదాహరణకు, cumVolumeని మునుపటి నిమిషం నుండి కాపీ చేయాలి మరియు మొదటిది 0కి సెట్ చేయాలి. డిక్షనరీ డేటాను ఉపయోగించి ఈ పారామితులన్నింటినీ సెట్ చేద్దాం రకం (రికార్డ్‌కి సారూప్యం):
// list ! list – создать словарь, 0n – float null, 0N – long null, `sym – тип символ, `sym1`sym2 – список символов
initWith:`sym`time`high`low`firstPrice`lastPrice`firstSize`lastSize`numTrades`volume`pvolume`turnover`avgPrice`avgSize`vwap`cumVolume!(`;00:00;0n;0n;0n;0n;0N;0N;0;0;0.0;0.0;0n;0n;0n;0);
aggCols:reverse key[initWith] except `sym`time; // список всех вычисляемых колонок, reverse объяснен ниже
నేను సౌలభ్యం కోసం నిఘంటువుకి సిమ్ మరియు సమయాన్ని జోడించాను, ఇప్పుడు initWith అనేది తుది సమగ్ర పట్టిక నుండి సిద్ధంగా ఉన్న లైన్, ఇక్కడ సరైన సిమ్ మరియు సమయాన్ని సెట్ చేయడానికి మిగిలి ఉంది. పట్టికకు కొత్త అడ్డు వరుసలను జోడించడానికి మీరు దీన్ని ఉపయోగించవచ్చు.
అగ్రిగేషన్ ఫంక్షన్‌ను సృష్టించేటప్పుడు మాకు aggCols అవసరం. Qలోని వ్యక్తీకరణలు (కుడి నుండి ఎడమకు) మూల్యాంకనం చేయబడే క్రమం కారణంగా జాబితా తప్పనిసరిగా విలోమం చేయబడాలి. కొన్ని నిలువు వరుసలు మునుపటి వాటిపై ఆధారపడి ఉన్నందున, గణన అధిక స్థాయి నుండి కమ్‌వాల్యూమ్‌కి వెళ్లేలా చేయడం లక్ష్యం.
మునుపటి దాని నుండి కొత్త నిమిషానికి కాపీ చేయవలసిన నిలువు వరుసలు, సౌలభ్యం కోసం సిమ్ కాలమ్ జోడించబడింది:
rollColumns:`sym`cumVolume;
ఇప్పుడు వాటిని ఎలా అప్‌డేట్ చేయాలి అనే దాని ప్రకారం నిలువు వరుసలను సమూహాలుగా విభజిద్దాము. మూడు రకాలను వేరు చేయవచ్చు:
- అక్యుమ్యులేటర్లు (వాల్యూమ్, టర్నోవర్,..) - మేము తప్పనిసరిగా మునుపటి విలువకు ఇన్‌కమింగ్ విలువను జోడించాలి.
- ప్రత్యేక పాయింట్‌తో (అధిక, తక్కువ, ..) - నిమిషంలో మొదటి విలువ ఇన్‌కమింగ్ డేటా నుండి తీసుకోబడుతుంది, మిగిలినవి ఫంక్షన్‌ని ఉపయోగించి లెక్కించబడతాయి.
- విశ్రాంతి. ఎల్లప్పుడూ ఫంక్షన్ ఉపయోగించి లెక్కించబడుతుంది.
ఈ తరగతులకు వేరియబుల్స్‌ని నిర్వచిద్దాం:
accumulatorCols:`numTrades`volume`pvolume`turnover;
specialCols:`high`low`firstPrice`firstSize;
గణన క్రమం
మేము సమగ్ర పట్టికను రెండు దశల్లో అప్‌డేట్ చేస్తాము. సామర్థ్యం కోసం, మేము మొదట ఇన్‌కమింగ్ టేబుల్‌ను కుదిస్తాము, తద్వారా ప్రతి అక్షరం మరియు నిమిషానికి ఒక వరుస మాత్రమే ఉంటుంది. మా అన్ని ఫంక్షన్లు ఇంక్రిమెంటల్ మరియు అసోసియేటివ్ అనే వాస్తవం ఈ అదనపు దశ యొక్క ఫలితం మారదని హామీ ఇస్తుంది. మీరు ఎంపికను ఉపయోగించి పట్టికను కుదించవచ్చు:
select high:max price, low:min price … by sym,time.minute from table
ఈ పద్ధతికి ప్రతికూలత ఉంది - లెక్కించిన నిలువు వరుసల సమితి ముందే నిర్వచించబడింది. అదృష్టవశాత్తూ, Qలో, మీరు డైనమిక్‌గా సృష్టించిన ఆర్గ్యుమెంట్‌లను భర్తీ చేయగల ఫంక్షన్‌గా కూడా సెలెక్ట్ చేయబడుతుంది:
?[table;whereClause;byClause;selectClause]
నేను ఆర్గ్యుమెంట్‌ల ఆకృతిని వివరంగా వివరించను; మా విషయంలో, ఎక్స్‌ప్రెషన్‌ల ద్వారా మరియు ఎంపిక చేసుకోవడం మాత్రమే నాన్‌ట్రివియల్‌గా ఉంటుంది మరియు అవి ఫారమ్ నిలువు! వ్యక్తీకరణల నిఘంటువులుగా ఉండాలి. అందువలన, కుదించే ఫంక్షన్ క్రింది విధంగా నిర్వచించవచ్చు:
selExpression:`high`low`firstPrice`lastPrice`firstSize`lastSize`numTrades`volume`pvolume`turnover!parse each ("max price";"min price";"first price";"last price";"first size";"last size";"count i";"sum size";"sum price";"sum price*size"); // each это функция map в Q для одного списка
preprocess:?[;();`sym`time!`sym`time.minute;selExpression];
స్పష్టత కోసం, నేను పార్స్ ఫంక్షన్‌ని ఉపయోగించాను, ఇది Q ఎక్స్‌ప్రెషన్‌తో స్ట్రింగ్‌ని ఎవాల్ ఫంక్షన్‌కి పంపగలిగే విలువగా మారుస్తుంది మరియు ఇది ఫంక్షన్ సెలెక్ట్‌లో అవసరం. ప్రిప్రాసెస్ అనేది ఎంచుకున్న ఫంక్షన్ యొక్క ప్రొజెక్షన్ (అంటే, పాక్షికంగా నిర్వచించబడిన ఆర్గ్యుమెంట్‌లతో కూడిన ఫంక్షన్)గా నిర్వచించబడిందని గమనించండి, ఒక ఆర్గ్యుమెంట్ (టేబుల్) లేదు. మనం టేబుల్‌కి ప్రిప్రాసెస్‌ని వర్తింపజేస్తే, మనకు కంప్రెస్డ్ టేబుల్ వస్తుంది.
రెండవ దశ సమగ్ర పట్టికను నవీకరించడం. ముందుగా సూడోకోడ్‌లో అల్గోరిథం వ్రాస్దాం:
for each sym in inputTable
idx: row index in agg table for sym+currentTime;
aggTable[idx;`high]: aggTable[idx;`high] | inputTable[sym;`high];
aggTable[idx;`volume]: aggTable[idx;`volume] + inputTable[sym;`volume];
…
Qలో, లూప్‌లకు బదులుగా మ్యాప్/రిడ్యూస్ ఫంక్షన్లను ఉపయోగించడం సర్వసాధారణం. అయితే Q అనేది వెక్టార్ లాంగ్వేజ్ మరియు మేము అన్ని ఆపరేషన్‌లను ఒకేసారి అన్ని చిహ్నాలకు సులభంగా వర్తింపజేయవచ్చు, ఆపై మొదటి ఉజ్జాయింపుకు మనం లూప్ లేకుండా చేయవచ్చు, అన్ని చిహ్నాలపై ఒకేసారి ఆపరేషన్‌లు చేయవచ్చు:
idx:calcIdx inputTable;
row:aggTable idx;
aggTable[idx;`high]: row[`high] | inputTable`high;
aggTable[idx;`volume]: row[`volume] + inputTable`volume;
…
కానీ మనం మరింత ముందుకు వెళ్ళవచ్చు, Q కి ప్రత్యేకమైన మరియు అత్యంత శక్తివంతమైన ఆపరేటర్ ఉంది - సాధారణీకరించిన అసైన్‌మెంట్ ఆపరేటర్. ఇది సూచికలు, విధులు మరియు వాదనల జాబితాను ఉపయోగించి సంక్లిష్ట డేటా నిర్మాణంలో విలువల సమితిని మార్చడానికి మిమ్మల్ని అనుమతిస్తుంది. మా విషయంలో ఇది ఇలా కనిపిస్తుంది:
idx:calcIdx inputTable;
rows:aggTable idx;
// .[target;(idx0;idx1;..);function;argument] ~ target[idx 0;idx 1;…]: function[target[idx 0;idx 1;…];argument], в нашем случае функция – это присваивание
.[aggTable;(idx;aggCols);:;flip (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
దురదృష్టవశాత్తూ, టేబుల్‌కి కేటాయించడానికి మీకు అడ్డు వరుసల జాబితా అవసరం, నిలువు వరుసలు కాదు మరియు మీరు ఫ్లిప్ ఫంక్షన్‌ని ఉపయోగించి మ్యాట్రిక్స్‌ను (వరుసల జాబితాకు నిలువు వరుసల జాబితా) బదిలీ చేయాలి. పెద్ద పట్టిక కోసం ఇది ఖరీదైనది, కాబట్టి బదులుగా మేము మ్యాప్ ఫంక్షన్‌ని ఉపయోగించి ప్రతి నిలువు వరుసకు సాధారణీకరించిన అసైన్‌మెంట్‌ని వర్తింపజేస్తాము (ఇది అపోస్ట్రోఫీలా కనిపిస్తుంది):
.[aggTable;;:;]'[(idx;)each aggCols; (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
మేము మళ్ళీ ఫంక్షన్ ప్రొజెక్షన్‌ని ఉపయోగిస్తాము. Qలో, జాబితాను సృష్టించడం కూడా ఒక ఫంక్షన్ అని కూడా గమనించండి మరియు జాబితాల జాబితాను పొందడానికి ప్రతి(మ్యాప్) ఫంక్షన్‌ని ఉపయోగించి మనం కాల్ చేయవచ్చు.
లెక్కించిన నిలువు వరుసల సెట్ స్థిరంగా లేదని నిర్ధారించుకోవడానికి, మేము పై వ్యక్తీకరణను డైనమిక్‌గా సృష్టిస్తాము. సంకలనం మరియు ఇన్‌పుట్ డేటాను సూచించడానికి అడ్డు వరుస మరియు inp వేరియబుల్‌లను ఉపయోగించి, ప్రతి నిలువు వరుసను లెక్కించడానికి మొదట ఫంక్షన్లను నిర్వచిద్దాం:
aggExpression:`high`low`firstPrice`lastPrice`firstSize`lastSize`avgPrice`avgSize`vwap`cumVolume!
("row[`high]|inp`high";"row[`low]&inp`low";"row`firstPrice";"inp`lastPrice";"row`firstSize";"inp`lastSize";"pvolume%numTrades";"volume%numTrades";"turnover%volume";"row[`cumVolume]+inp`volume");
కొన్ని నిలువు వరుసలు ప్రత్యేకమైనవి; వాటి మొదటి విలువ ఫంక్షన్ ద్వారా లెక్కించబడకూడదు. అడ్డు వరుస[`numTrades] నిలువు వరుస ద్వారా ఇది మొదటిదని మేము గుర్తించగలము - అది 0ని కలిగి ఉంటే, అప్పుడు విలువ మొదటిది. Qకి ఎంపిక చేయబడిన ఫంక్షన్ ఉంది - ?[బూలియన్ జాబితా;జాబితా1;జాబితా2] - ఇది మొదటి ఆర్గ్యుమెంట్‌లోని షరతుపై ఆధారపడి జాబితా 1 లేదా 2 నుండి విలువను ఎంచుకుంటుంది:
// high -> ?[isFirst;inp`high;row[`high]|inp`high]
// @ - тоже обобщенное присваивание для случая когда индекс неглубокий
@[`aggExpression;specialCols;{[x;y]"?[isFirst;inp`",y,";",x,"]"};string specialCols];
ఇక్కడ నేను నా ఫంక్షన్‌తో సాధారణీకరించిన అసైన్‌మెంట్‌ని పిలిచాను (కర్లీ బ్రేస్‌లలోని వ్యక్తీకరణ). ఇది ప్రస్తుత విలువ (మొదటి ఆర్గ్యుమెంట్) మరియు నేను 4వ పరామితిలో పాస్ చేసే అదనపు ఆర్గ్యుమెంట్‌ని అందుకుంటుంది.
బ్యాటరీ స్పీకర్‌లను విడిగా జోడిద్దాం, ఎందుకంటే వాటి ఫంక్షన్ ఒకే విధంగా ఉంటుంది:
// volume -> row[`volume]+inp`volume
aggExpression[accumulatorCols]:{"row[`",x,"]+inp`",x } each string accumulatorCols;
ఇది Q ప్రమాణాల ప్రకారం సాధారణ అసైన్‌మెంట్, కానీ నేను ఒకేసారి విలువల జాబితాను కేటాయిస్తున్నాను. చివరగా, ప్రధాన విధిని సృష్టిద్దాం:
// ":",/:aggExprs ~ map[{":",x};aggExpr] => ":row[`high]|inp`high" присвоим вычисленное значение переменной, потому что некоторые колонки зависят от уже вычисленных значений
// string[cols],'exprs ~ map[,;string[cols];exprs] => "high:row[`high]|inp`high" завершим создание присваивания. ,’ расшифровывается как map[concat]
// ";" sv exprs – String from Vector (sv), соединяет список строк вставляя “;” посредине
updateAgg:value "{[aggTable;idx;inp] row:aggTable idx; isFirst_0=row`numTrades; .[aggTable;;:;]'[(idx;)each aggCols;(",(";"sv string[aggCols],'":",/:aggExpression aggCols),")]}";
ఈ వ్యక్తీకరణతో, నేను పైన ఇచ్చిన ఎక్స్‌ప్రెషన్‌ను కలిగి ఉన్న స్ట్రింగ్ నుండి డైనమిక్‌గా ఫంక్షన్‌ని క్రియేట్ చేస్తాను. ఫలితం ఇలా కనిపిస్తుంది:
{[aggTable;idx;inp] rows:aggTable idx; isFirst_0=row`numTrades; .[aggTable;;:;]'[(idx;)each aggCols ;(cumVolume:row[`cumVolume]+inp`cumVolume;… ; high:?[isFirst;inp`high;row[`high]|inp`high])]}
Qలో మూల్యాంకన క్రమం కుడి నుండి ఎడమకు ఉన్నందున నిలువు మూల్యాంకన క్రమం విలోమం చేయబడింది.
ఇప్పుడు మనకు గణనలకు అవసరమైన రెండు ప్రధాన విధులు ఉన్నాయి, మేము కొద్దిగా మౌలిక సదుపాయాలను జోడించాలి మరియు సేవ సిద్ధంగా ఉంది.
చివరి దశలు
మేము అన్ని పనులను చేసే ప్రిప్రాసెస్ మరియు అప్‌డేట్Agg ఫంక్షన్లను కలిగి ఉన్నాము. కానీ నిమిషాల ద్వారా సరైన పరివర్తనను నిర్ధారించడం మరియు అగ్రిగేషన్ కోసం సూచికలను లెక్కించడం ఇప్పటికీ అవసరం. ముందుగా, init ఫంక్షన్‌ని నిర్వచిద్దాం:
init:{
tradeAgg:: 0#enlist[initWith]; // создаем пустую типизированную таблицу, enlist превращает словарь в таблицу, а 0# означает взять 0 элементов из нее
currTime::00:00; // начнем с 0, :: означает, что присваивание в глобальную переменную
currSyms::`u#`symbol$(); // `u# - превращает список в дерево, для ускорения поиска элементов
offset::0; // индекс в tradeAgg, где начинается текущая минута
rollCache:: `sym xkey update `u#sym from rollColumns#tradeAgg; // кэш для последних значений roll колонок, таблица с ключом sym
}
మేము రోల్ ఫంక్షన్‌ను కూడా నిర్వచిస్తాము, ఇది ప్రస్తుత నిమిషాన్ని మారుస్తుంది:
roll:{[tm]
if[currTime>tm; :init[]]; // если перевалили за полночь, то просто вызовем init
rollCache,::offset _ rollColumns#tradeAgg; // обновим кэш – взять roll колонки из aggTable, обрезать, вставить в rollCache
offset::count tradeAgg;
currSyms::`u#`$();
}
కొత్త అక్షరాలను జోడించడానికి మాకు ఒక ఫంక్షన్ అవసరం:
addSyms:{[syms]
currSyms,::syms; // добавим в список известных
// добавим в таблицу sym, time и rollColumns воспользовавшись обобщенным присваиванием.
// Функция ^ подставляет значения по умолчанию для roll колонок, если символа нет в кэше. value flip table возвращает список колонок в таблице.
`tradeAgg upsert @[count[syms]#enlist initWith;`sym`time,cols rc;:;(syms;currTime), (initWith cols rc)^value flip rc:rollCache ([] sym: syms)];
}
చివరకు, అప్‌డిట్ ఫంక్షన్ (Q సేవల కోసం ఈ ఫంక్షన్‌కి సాంప్రదాయ పేరు), ఇది డేటాను జోడించడానికి క్లయింట్ ద్వారా పిలువబడుతుంది:
upd:{[tblName;data] // tblName нам не нужно, но обычно сервис обрабатывает несколько таблиц
tm:exec distinct time from data:() xkey preprocess data; // preprocess & calc time
updMinute[data] each tm; // добавим данные для каждой минуты
};
updMinute:{[data;tm]
if[tm<>currTime; roll tm; currTime::tm]; // поменяем минуту, если необходимо
data:select from data where time=tm; // фильтрация
if[count msyms:syms where not (syms:data`sym)in currSyms; addSyms msyms]; // новые символы
updateAgg[`tradeAgg;offset+currSyms?syms;data]; // обновим агрегированную таблицу. Функция ? ищет индекс элементов списка справа в списке слева.
};
అంతే. వాగ్దానం చేసినట్లుగా మా సేవ యొక్క పూర్తి కోడ్ ఇక్కడ ఉంది, కొన్ని పంక్తులు:
initWith:`sym`time`high`low`firstPrice`lastPrice`firstSize`lastSize`numTrades`volume`pvolume`turnover`avgPrice`avgSize`vwap`cumVolume!(`;00:00;0n;0n;0n;0n;0N;0N;0;0;0.0;0.0;0n;0n;0n;0);
aggCols:reverse key[initWith] except `sym`time;
rollColumns:`sym`cumVolume;
accumulatorCols:`numTrades`volume`pvolume`turnover;
specialCols:`high`low`firstPrice`firstSize;
selExpression:`high`low`firstPrice`lastPrice`firstSize`lastSize`numTrades`volume`pvolume`turnover!parse each ("max price";"min price";"first price";"last price";"first size";"last size";"count i";"sum size";"sum price";"sum price*size");
preprocess:?[;();`sym`time!`sym`time.minute;selExpression];
aggExpression:`high`low`firstPrice`lastPrice`firstSize`lastSize`avgPrice`avgSize`vwap`cumVolume!("row[`high]|inp`high";"row[`low]&inp`low";"row`firstPrice";"inp`lastPrice";"row`firstSize";"inp`lastSize";"pvolume%numTrades";"volume%numTrades";"turnover%volume";"row[`cumVolume]+inp`volume");
@[`aggExpression;specialCols;{"?[isFirst;inp`",y,";",x,"]"};string specialCols];
aggExpression[accumulatorCols]:{"row[`",x,"]+inp`",x } each string accumulatorCols;
updateAgg:value "{[aggTable;idx;inp] row:aggTable idx; isFirst_0=row`numTrades; .[aggTable;;:;]'[(idx;)each aggCols;(",(";"sv string[aggCols],'":",/:aggExpression aggCols),")]}"; / '
init:{
tradeAgg::0#enlist[initWith];
currTime::00:00;
currSyms::`u#`symbol$();
offset::0;
rollCache:: `sym xkey update `u#sym from rollColumns#tradeAgg;
};
roll:{[tm]
if[currTime>tm; :init[]];
rollCache,::offset _ rollColumns#tradeAgg;
offset::count tradeAgg;
currSyms::`u#`$();
};
addSyms:{[syms]
currSyms,::syms;
`tradeAgg upsert @[count[syms]#enlist initWith;`sym`time,cols rc;:;(syms;currTime),(initWith cols rc)^value flip rc:rollCache ([] sym: syms)];
};
upd:{[tblName;data] updMinute[data] each exec distinct time from data:() xkey preprocess data};
updMinute:{[data;tm]
if[tm<>currTime; roll tm; currTime::tm];
data:select from data where time=tm;
if[count msyms:syms where not (syms:data`sym)in currSyms; addSyms msyms];
updateAgg[`tradeAgg;offset+currSyms?syms;data];
};
పరీక్ష
సేవ యొక్క పనితీరును తనిఖీ చేద్దాం. దీన్ని చేయడానికి, దానిని ప్రత్యేక ప్రక్రియలో అమలు చేద్దాం (కోడ్‌ను service.q ఫైల్‌లో ఉంచండి) మరియు init ఫంక్షన్‌కు కాల్ చేయండి:
q service.q –p 5566
q)init[]
మరొక కన్సోల్‌లో, రెండవ Q ప్రక్రియను ప్రారంభించి, మొదటి దానికి కనెక్ట్ చేయండి:
h:hopen `:host:5566
h:hopen 5566 // если оба на одном хосте
ముందుగా, చిహ్నాల జాబితాను రూపొందించండి - 10000 ముక్కలు మరియు యాదృచ్ఛిక పట్టికను సృష్టించడానికి ఒక ఫంక్షన్‌ను జోడించండి. రెండవ కన్సోల్‌లో:
syms:`IBM`AAPL`GOOG,-9997?`8
rnd:{[n;t] ([] sym:n?syms; time:t+asc n#til 25; price:n?10f; size:n?10)}
పట్టికలో వాటి కోసం వెతకడం సులభతరం చేయడానికి నేను జాబితాకు మూడు నిజమైన చిహ్నాలను జోడించాను. Rnd ఫంక్షన్ n అడ్డు వరుసలతో యాదృచ్ఛిక పట్టికను సృష్టిస్తుంది, ఇక్కడ సమయం t నుండి t+25 మిల్లీసెకన్ల వరకు మారుతుంది.
ఇప్పుడు మీరు సేవకు డేటాను పంపడానికి ప్రయత్నించవచ్చు (మొదటి పది గంటలను జోడించండి):
{h (`upd;`trade;rnd[10000;x])} each `time$00:00 + til 60*10
టేబుల్ అప్‌డేట్ చేయబడిందని మీరు సేవలో తనిఖీ చేయవచ్చు:
c 25 200
select from tradeAgg where sym=`AAPL
-20#select from tradeAgg where sym=`AAPL
ఫలితంగా:
sym|time|high|low|firstPrice|lastPrice|firstSize|lastSize|numTrades|volume|pvolume|turnover|avgPrice|avgSize|vwap|cumVolume
--|--|--|--|--|--------------------------------
AAPL|09:27|9.258904|9.258904|9.258904|9.258904|8|8|1|8|9.258904|74.07123|9.258904|8|9.258904|2888
AAPL|09:28|9.068162|9.068162|9.068162|9.068162|7|7|1|7|9.068162|63.47713|9.068162|7|9.068162|2895
AAPL|09:31|4.680449|0.2011121|1.620827|0.2011121|1|5|4|14|9.569556|36.84342|2.392389|3.5|2.631673|2909
AAPL|09:33|2.812535|2.812535|2.812535|2.812535|6|6|1|6|2.812535|16.87521|2.812535|6|2.812535|2915
AAPL|09:34|5.099025|5.099025|5.099025|5.099025|4|4|1|4|5.099025|20.3961|5.099025|4|5.099025|2919సేవ నిమిషానికి ఎంత డేటాను ప్రాసెస్ చేయగలదో తెలుసుకోవడానికి ఇప్పుడు లోడ్ పరీక్షను చేద్దాం. మేము నవీకరణ విరామాన్ని 25 మిల్లీసెకన్లకు సెట్ చేసామని నేను మీకు గుర్తు చేస్తున్నాను. దీని ప్రకారం, డేటాను అభ్యర్థించడానికి వినియోగదారులకు సమయం ఇవ్వడానికి సేవ తప్పనిసరిగా (సగటున) ప్రతి నవీకరణకు కనీసం 20 మిల్లీసెకన్లకు సరిపోవాలి. రెండవ ప్రక్రియలో కింది వాటిని నమోదు చేయండి:
tm:10:00:00.000
stressTest:{[n] 1 string[tm]," "; times,::h ({st:.z.T; upd[`trade;x]; .z.T-st};rnd[n;tm]); tm+:25}
start:{[n] times::(); do[4800;stressTest[n]]; -1 " "; `min`avg`med`max!(min times;avg times;med times;max times)}
4800 అంటే రెండు నిమిషాలు. మీరు ప్రతి 1000 మిల్లీసెకన్లకు 25 అడ్డు వరుసల కోసం ముందుగా రన్ చేయడానికి ప్రయత్నించవచ్చు:
start 1000
నా విషయంలో, ప్రతి అప్‌డేట్‌కి దాదాపు రెండు మిల్లీసెకన్ల ఫలితం ఉంటుంది. కాబట్టి నేను వెంటనే అడ్డు వరుసల సంఖ్యను 10.000కి పెంచుతాను:
start 10000
ఫలితంగా:
min| 00:00:00.004
avg| 9.191458
med| 9f
max| 00:00:00.030
మళ్ళీ, ప్రత్యేకంగా ఏమీ లేదు, కానీ ఇది నిమిషానికి 24 మిలియన్ లైన్లు, సెకనుకు 400 వేలు. 25 మిల్లీసెకన్ల కంటే ఎక్కువ, నవీకరణ కేవలం 5 సార్లు మాత్రమే మందగించింది, స్పష్టంగా నిమిషం మారినప్పుడు. 100.000కి పెంచుదాం:
start 100000
ఫలితంగా:
min| 00:00:00.013
avg| 25.11083
med| 24f
max| 00:00:00.108
q)sum times
00:02:00.532
మీరు చూడగలిగినట్లుగా, సేవ కేవలం భరించవలసి ఉంటుంది, అయితే ఇది తేలుతూనే ఉంటుంది. అటువంటి డేటా పరిమాణం (నిమిషానికి 240 మిలియన్ వరుసలు) చాలా పెద్దది; అటువంటి సందర్భాలలో, సేవ యొక్క అనేక క్లోన్‌లను (లేదా డజన్ల కొద్దీ క్లోన్‌లు కూడా) ప్రారంభించడం సాధారణం, వీటిలో ప్రతి ఒక్కటి అక్షరాల్లో కొంత భాగాన్ని మాత్రమే ప్రాసెస్ చేస్తుంది. అయినప్పటికీ, ప్రాథమికంగా డేటా నిల్వపై దృష్టి సారించే అన్వయించబడిన భాష కోసం ఫలితం ఆకట్టుకుంటుంది.
ప్రతి నవీకరణ పరిమాణంతో సమయం ఎందుకు నాన్-లీనియర్‌గా పెరుగుతుంది అనే ప్రశ్న తలెత్తవచ్చు. కారణం ఏమిటంటే, ష్రింక్ ఫంక్షన్ వాస్తవానికి C ఫంక్షన్, ఇది updateAgg కంటే చాలా సమర్థవంతమైనది. నిర్దిష్ట నవీకరణ పరిమాణం (సుమారు 10.000) నుండి ప్రారంభించి, updateAgg దాని పైకప్పుకు చేరుకుంటుంది మరియు దాని అమలు సమయం నవీకరణ పరిమాణంపై ఆధారపడి ఉండదు. ప్రాథమిక దశ Q కారణంగా, అటువంటి డేటా వాల్యూమ్‌లను సేవ జీర్ణించుకోగలుగుతుంది. పెద్ద డేటాతో పని చేస్తున్నప్పుడు సరైన అల్గారిథమ్‌ను ఎంచుకోవడం ఎంత ముఖ్యమో ఇది హైలైట్ చేస్తుంది. మరొక పాయింట్ మెమరీలో డేటా యొక్క సరైన నిల్వ. డేటా నిలువుగా నిల్వ చేయబడకపోతే లేదా సమయానుసారంగా ఆర్డర్ చేయబడకపోతే, TLB కాష్ మిస్ వంటి వాటితో మనకు సుపరిచితం అవుతుంది - ప్రాసెసర్ చిరునామా కాష్‌లో మెమరీ పేజీ చిరునామా లేకపోవడం. చిరునామా కోసం శోధించడం విఫలమైతే దాదాపు 30 రెట్లు ఎక్కువ సమయం పడుతుంది మరియు డేటా చెల్లాచెదురుగా ఉంటే, అది సేవను అనేక సార్లు నెమ్మదిస్తుంది.
తీర్మానం
ఈ కథనంలో, KDB+ మరియు Q డేటాబేస్ పెద్ద డేటాను నిల్వ చేయడానికి మరియు ఎంపిక ద్వారా సులభంగా యాక్సెస్ చేయడానికి మాత్రమే కాకుండా, వందల మిలియన్ల వరుసలు/గిగాబైట్ల డేటాను జీర్ణం చేయగల డేటా ప్రాసెసింగ్ సేవలను రూపొందించడానికి కూడా సరిపోతుందని నేను చూపించాను. ఒకే Q ప్రక్రియ. Q భాష దాని వెక్టర్ స్వభావం, అంతర్నిర్మిత SQL మాండలికం ఇంటర్‌ప్రెటర్ మరియు చాలా విజయవంతమైన లైబ్రరీ ఫంక్షన్‌ల కారణంగా డేటా ప్రాసెసింగ్‌కు సంబంధించిన అల్గారిథమ్‌ల యొక్క అత్యంత సంక్షిప్త మరియు సమర్థవంతమైన అమలును అనుమతిస్తుంది.
పైన పేర్కొన్నది Q చేయగలిగిన దానిలో ఒక భాగమని నేను గమనిస్తాను, దీనికి ఇతర ప్రత్యేక లక్షణాలు కూడా ఉన్నాయి. ఉదాహరణకు, చాలా సులభమైన IPC ప్రోటోకాల్, ఇది వ్యక్తిగత Q ప్రక్రియల మధ్య సరిహద్దును చెరిపివేస్తుంది మరియు ప్రపంచంలోని వివిధ ప్రాంతాల్లోని డజన్ల కొద్దీ సర్వర్‌లలో ఉండే ఒకే నెట్‌వర్క్‌లో వందల కొద్దీ ఈ ప్రక్రియలను కలపడానికి మిమ్మల్ని అనుమతిస్తుంది.
మూలం: www.habr.com