Q და KDB+ ენის მახასიათებლები რეალურ დროში სერვისის მაგალითის გამოყენებით

თქვენ შეგიძლიათ წაიკითხოთ რა არის KDB+ ბაზა, Q პროგრამირების ენა, რა არის მათი ძლიერი და სუსტი მხარეები ჩემს წინა მუხლი და მოკლედ შესავალში. სტატიაში ჩვენ განვახორციელებთ სერვისს Q-ზე, რომელიც დაამუშავებს შემომავალ მონაცემთა ნაკადს და გამოთვლის სხვადასხვა აგრეგაციის ფუნქციებს ყოველ წუთში „რეალურ დროში“ რეჟიმში (ანუ მას ექნება დრო, რომ გამოთვალოს ყველაფერი მონაცემთა მომდევნო ნაწილამდე). Q-ის მთავარი მახასიათებელია ის, რომ ის არის ვექტორული ენა, რომელიც საშუალებას გაძლევთ იმუშაოთ არა ცალკეულ ობიექტებთან, არამედ მათი მასივებით, მასივების მასივებით და სხვა რთული ობიექტებით. ისეთი ენები, როგორიც არის Q და მისი ნათესავები K, J, APL, განთქმულია მათი ლაკონურობით. ხშირად, პროგრამა, რომელიც იკავებს კოდის რამდენიმე ეკრანს ნაცნობ ენაზე, როგორიცაა Java, შეიძლება დაიწეროს მათზე რამდენიმე სტრიქონში. ეს არის ის, რისი დემონსტრირებაც მსურს ამ სტატიაში.

შესავალი

KDB+ არის სვეტოვანი მონაცემთა ბაზა, რომელიც ორიენტირებულია მონაცემთა ძალიან დიდ რაოდენობაზე, დალაგებულია კონკრეტული გზით (ძირითადად დროის მიხედვით). იგი ძირითადად გამოიყენება ფინანსურ ინსტიტუტებში - ბანკებში, საინვესტიციო ფონდებში, სადაზღვევო კომპანიებში. Q ენა არის KDB+-ის შიდა ენა, რომელიც საშუალებას გაძლევთ ეფექტურად იმუშაოთ ამ მონაცემებთან. Q იდეოლოგია არის ლაკონურობა და ეფექტურობა, ხოლო სიცხადე ეწირება. ეს გამართლებულია იმით, რომ ვექტორული ენის გაგება ნებისმიერ შემთხვევაში რთული იქნება, ხოლო ჩანაწერის სიმოკლე და სიმდიდრე საშუალებას გაძლევთ ნახოთ პროგრამის გაცილებით დიდი ნაწილი ერთ ეკრანზე, რაც საბოლოოდ აადვილებს მის გაგებას.

ამ სტატიაში ჩვენ განვახორციელებთ სრულფასოვან პროგრამას Q-ში და შეიძლება გინდოდეთ სცადოთ იგი. ამისათვის დაგჭირდებათ ფაქტობრივი Q. შეგიძლიათ ჩამოტვირთოთ უფასო 32-ბიტიანი ვერსია kx კომპანიის ვებსაიტზე – www.kx.com. იქ, თუ დაგაინტერესებთ, იპოვით საცნობარო ინფორმაციას Q, წიგნზე Q მოკვდავებისთვის და სხვადასხვა სტატიები ამ თემაზე.

პრობლემის შესახებ განცხადება

არის წყარო, რომელიც აგზავნის ცხრილს მონაცემებით ყოველ 25 მილიწამში. ვინაიდან KDB+ ძირითადად გამოიყენება ფინანსებში, ჩვენ ვივარაუდებთ, რომ ეს არის ტრანზაქციების ცხრილი (ვაჭრობა), რომელსაც აქვს შემდეგი სვეტები: დრო (დრო მილიწამებში), sym (კომპანიის აღნიშვნა საფონდო ბირჟაზე - IBM, AAPL,…), ფასი (ფასი, რომლითაც შეიძინა აქციები), ზომა (ტრანზაქციის ზომა). 25 მილიწამიანი ინტერვალი თვითნებურია, არც ისე მცირე და არც ძალიან გრძელი. მისი არსებობა ნიშნავს, რომ მონაცემები სერვისში უკვე ბუფერირებულია. მარტივი იქნება ბუფერირების დანერგვა სერვისის მხარეს, მათ შორის დინამიური ბუფერირება მიმდინარე დატვირთვის მიხედვით, მაგრამ სიმარტივისთვის ჩვენ ყურადღებას გავამახვილებთ ფიქსირებულ ინტერვალზე.

სერვისმა უნდა დაითვალოს ყოველ წუთს sym სვეტიდან თითოეული შემომავალი სიმბოლოსთვის ფუნქციების დაგროვების ნაკრები - მაქსიმალური ფასი, საშუალო ფასი, ჯამის ზომა და ა.შ. გამოსადეგი ინფორმაცია. სიმარტივისთვის, ჩვენ ვივარაუდებთ, რომ ყველა ფუნქციის გამოთვლა შესაძლებელია თანდათანობით, ე.ი. ახალი მნიშვნელობის მისაღებად, საკმარისია იცოდეთ ორი რიცხვი - ძველი და შემომავალი მნიშვნელობები. მაგალითად, ფუნქციებს max, საშუალო, sum აქვს ეს თვისება, მაგრამ მედიანა ფუნქციას არა.

ჩვენ ასევე ვივარაუდებთ, რომ შემომავალი მონაცემთა ნაკადი არის დროში შეკვეთილი. ეს მოგვცემს შესაძლებლობას ვიმუშაოთ მხოლოდ ბოლო წუთთან. პრაქტიკაში საკმარისია მიმდინარე და წინა წუთებთან მუშაობა, თუ რაიმე განახლება დაგვიანებულია. სიმარტივისთვის, ჩვენ არ განვიხილავთ ამ შემთხვევას.

აგრეგაციის ფუნქციები

საჭირო აგრეგაციის ფუნქციები ჩამოთვლილია ქვემოთ. რაც შეიძლება მეტი მათგანი ავიღე სერვისზე დატვირთვის გასაზრდელად:

• მაღალი – მაქსიმალური ფასი – მაქსიმალური ფასი წუთში.
• დაბალი – მინიმალური ფასი – მინიმალური ფასი წუთში.
• პირველი ფასი – პირველი ფასი – პირველი ფასი წუთში.
• lastPrice – ბოლო ფასი – ბოლო ფასი წუთში.
• firstSize – პირველი ზომა – პირველი ვაჭრობის ზომა წუთში.
• lastSize - ბოლო ზომა - ბოლო ვაჭრობის ზომა წუთში.
• numTrades – count i – გარიგებების რაოდენობა წუთში.
• მოცულობა – ჯამის ზომა – ვაჭრობის ზომის ჯამი წუთში.
• pvolume – ჯამის ფასი – ფასების ჯამი წუთში, საჭიროა avgPrice-სთვის.
• – ჯამური ბრუნვის ფასი*ზომა – ტრანზაქციების ჯამური მოცულობა წუთში.
• avgPrice – pvolume%numTrades – საშუალო ფასი წუთში.
• avgSize – მოცულობა%numTrades – ვაჭრობის საშუალო ზომა წუთში.
• vwap – ბრუნვა%მოცულობა – საშუალო ფასი წუთში ტრანზაქციის ზომის მიხედვით.
• cumVolume – ჯამის მოცულობა – ტრანზაქციების დაგროვილი ზომა მთელი დროის განმავლობაში.

მოდით დაუყოვნებლივ განვიხილოთ ერთი არააშკარა პუნქტი - როგორ მოვახდინოთ ამ სვეტების ინიციალიზაცია პირველად და ყოველი მომდევნო წუთისთვის. პირველიPrice ტიპის ზოგიერთი სვეტი ყოველ ჯერზე უნდა იყოს ინიციალიზებული ნულამდე; მათი მნიშვნელობა განუსაზღვრელია. მოცულობის სხვა ტიპები ყოველთვის უნდა იყოს 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 объяснен ниже

მოხერხებულობისთვის ლექსიკონს დავამატე sym და დრო, ახლა initWith არის მზა სტრიქონი საბოლოო აგრეგირებული ცხრილიდან, სადაც რჩება სწორი სიმბოლოს და დროის დაყენება. თქვენ შეგიძლიათ გამოიყენოთ იგი ცხრილში ახალი სტრიქონების დასამატებლად.

აგრეგაციის ფუნქციის შექმნისას დაგვჭირდება aggCols. სია უნდა იყოს შებრუნებული Q-ში გამოსახულებების შეფასების თანმიმდევრობის გამო (მარჯვნიდან მარცხნივ). მიზანია უზრუნველყოს, რომ გაანგარიშება გადავიდეს მაღალიდან cumVolume-მდე, რადგან ზოგიერთი სვეტი დამოკიდებულია წინა სვეტებზე.

სვეტები, რომლებიც უნდა დაკოპირდეს წინა წუთში ახალ წუთში, მოხერხებულობისთვის ემატება sym სვეტი:

rollColumns:`sym`cumVolume;

ახლა მოდით დავყოთ სვეტები ჯგუფებად იმის მიხედვით, თუ როგორ უნდა განახლდეს. შეიძლება განვასხვავოთ სამი ტიპი:

1. აკუმულატორები (მოცულობა, ბრუნვა,..) – შემომავალი მნიშვნელობა უნდა დავუმატოთ წინას.
2. სპეციალური წერტილით (მაღალი, დაბალი, ..) – წუთში პირველი მნიშვნელობა აღებულია შემოსული მონაცემებიდან, დანარჩენი გამოითვლება ფუნქციის გამოყენებით.
3. დაისვენე. ყოველთვის გამოითვლება ფუნქციის გამოყენებით.

მოდით განვსაზღვროთ ცვლადები ამ კლასებისთვის:

accumulatorCols:`numTrades`volume`pvolume`turnover;
specialCols:`high`low`firstPrice`firstSize;

გაანგარიშების ბრძანება

ჩვენ განვაახლებთ საერთო ცხრილს ორ ეტაპად. ეფექტურობისთვის, ჩვენ ჯერ შევამცირებთ შემომავალ ცხრილს ისე, რომ თითოეულ სიმბოლოსა და წუთში იყოს მხოლოდ ერთი მწკრივი. ის ფაქტი, რომ ჩვენი ყველა ფუნქცია არის დამატებითი და ასოციაციური, გარანტიას იძლევა, რომ ამ დამატებითი ნაბიჯის შედეგი არ შეიცვლება. ცხრილის შემცირება შეგიძლიათ შერჩევის გამოყენებით:

select high:max price, low:min price … by sym,time.minute from table

ამ მეთოდს აქვს მინუსი - გამოთვლილი სვეტების ნაკრები წინასწარ არის განსაზღვრული. საბედნიეროდ, Q-ში select ასევე განხორციელებულია როგორც ფუნქცია, სადაც შეგიძლიათ ჩაანაცვლოთ დინამიურად შექმნილი არგუმენტები:

?[table;whereClause;byClause;selectClause]

არგუმენტების ფორმატს დეტალურად არ აღვწერ, ჩვენს შემთხვევაში მხოლოდ გამონათქვამები by და select იქნება არატრივიალური და ისინი უნდა იყოს ფორმის სვეტების!გამონათქვამების ლექსიკონები. ამრიგად, შემცირების ფუნქცია შეიძლება განისაზღვროს შემდეგნაირად:

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 გამოსახულებით სტრიქონს მნიშვნელობად, რომელიც შეიძლება გადაეცეს eval ფუნქციას და რომელიც საჭიროა ფუნქციის შერჩევაში. ასევე გაითვალისწინეთ, რომ წინასწარი პროცესი განისაზღვრება, როგორც პროექცია (ანუ ფუნქცია ნაწილობრივ განსაზღვრული არგუმენტებით) შერჩეული ფუნქციის, ერთი არგუმენტი (ცხრილი) აკლია. თუ ცხრილზე გამოვიყენებთ წინასწარ პროცესს, მივიღებთ შეკუმშულ ცხრილს.

მეორე ეტაპი არის აგრეგირებული ცხრილის განახლება. მოდით, ჯერ დავწეროთ ალგორითმი ფსევდოკოდში:

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;…)];

სამწუხაროდ, ცხრილისთვის მინიჭებისთვის საჭიროა სტრიქონების სია და არა სვეტები და თქვენ უნდა გადაიტანოთ მატრიცა (სვეტების სია მწკრივების სიაში) Flip ფუნქციის გამოყენებით. ეს ძვირია დიდი ცხრილისთვის, ამიტომ ჩვენ ვიყენებთ განზოგადებულ დავალებას თითოეულ სვეტზე ცალ-ცალკე რუკის ფუნქციის გამოყენებით (რომელიც აპოსტროფს ჰგავს):

.[aggTable;;:;]'[(idx;)each aggCols; (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];

ჩვენ კვლავ ვიყენებთ ფუნქციის პროექციას. ასევე გაითვალისწინეთ, რომ Q-ში სიის შექმნა ასევე ფუნქციაა და შეგვიძლია მას ვუწოდოთ ყოველი(map) ფუნქციის გამოყენებით სიების სიის მისაღებად.

იმის უზრუნველსაყოფად, რომ გამოთვლილი სვეტების ნაკრები არ არის დაფიქსირებული, ჩვენ შევქმნით ზემოხსენებულ გამონათქვამს დინამიურად. მოდით, ჯერ განვსაზღვროთ ფუნქციები თითოეული სვეტის გამოსათვლელად, მწკრივის და 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-ს აქვს არჩევის ფუნქცია - ?[Boolean list;list1;list2] - რომელიც ირჩევს მნიშვნელობას 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-ში შეფასების თანმიმდევრობა არის მარჯვნიდან მარცხნივ.

ახლა ჩვენ გვაქვს ორი ძირითადი ფუნქცია, რომელიც აუცილებელია გამოთვლებისთვის, უბრალოდ უნდა დავამატოთ მცირე ინფრასტრუქტურა და სერვისი მზად არის.

საბოლოო ნაბიჯები

ჩვენ გვაქვს წინასწარი დამუშავების და განახლების ფუნქციები, რომლებიც ასრულებს ყველა სამუშაოს. მაგრამ მაინც აუცილებელია წუთებში სწორი გადასვლის უზრუნველსაყოფად და აგრეგაციის ინდექსების გამოთვლა. უპირველეს ყოვლისა, მოდით განვსაზღვროთ 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)];
 }

და ბოლოს, upd ფუნქცია (ამ ფუნქციის ტრადიციული სახელი 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 მილიონი მწკრივი) ძალიან დიდია; ასეთ შემთხვევებში ჩვეულებრივია სერვისის რამდენიმე კლონის (ან თუნდაც ათობით კლონის) გაშვება, რომელთაგან თითოეული ამუშავებს პერსონაჟების მხოლოდ ნაწილს. და მაინც, შედეგი შთამბეჭდავია ინტერპრეტაციული ენისთვის, რომელიც ძირითადად ყურადღებას ამახვილებს მონაცემთა შენახვაზე.

შეიძლება გაჩნდეს კითხვა, რატომ იზრდება დრო არაწრფივად თითოეული განახლების ზომით. მიზეზი ის არის, რომ shrink ფუნქცია რეალურად არის C ფუნქცია, რომელიც ბევრად უფრო ეფექტურია ვიდრე updateAgg. განახლების გარკვეული ზომიდან დაწყებული (დაახლოებით 10.000), updateAgg აღწევს თავის ჭერს და შემდეგ მისი შესრულების დრო არ არის დამოკიდებული განახლების ზომაზე. ეს არის Q წინასწარი ნაბიჯის გამო, რომ სერვისს შეუძლია მონაცემთა ასეთი მოცულობის დამუშავება. ეს ხაზს უსვამს იმას, თუ რამდენად მნიშვნელოვანია სწორი ალგორითმის არჩევა დიდ მონაცემებთან მუშაობისას. კიდევ ერთი წერტილი არის მონაცემთა სწორად შენახვა მეხსიერებაში. თუ მონაცემები არ იქნებოდა შენახული სვეტურად ან არ იყო შეკვეთილი დროის მიხედვით, მაშინ ჩვენ გავეცანით ისეთ რამეს, როგორიცაა TLB ქეშის გამოტოვება - მეხსიერების გვერდის მისამართის არარსებობა პროცესორის მისამართების ქეშში. წარუმატებლობის შემთხვევაში მისამართის ძიებას დაახლოებით 30-ჯერ მეტი დრო სჭირდება, ხოლო თუ მონაცემები გაფანტულია, მას შეუძლია რამდენჯერმე შეანელოს სერვისი.

დასკვნა

ამ სტატიაში მე ვაჩვენე, რომ KDB+ და Q მონაცემთა ბაზა შესაფერისია არა მხოლოდ დიდი მონაცემების შესანახად და მასზე წვდომისთვის არჩევის საშუალებით, არამედ მონაცემთა დამუშავების სერვისების შესაქმნელად, რომლებსაც შეუძლიათ ასობით მილიონი სტრიქონი/გიგაბაიტი მონაცემების დამუშავება. ერთი Q პროცესი. Q ენა თავისთავად იძლევა მონაცემთა დამუშავებასთან დაკავშირებული ალგორითმების უკიდურესად ლაკონურ და ეფექტურ განხორციელების საშუალებას მისი ვექტორული ბუნების, ჩაშენებული SQL დიალექტური თარჯიმანისა და ბიბლიოთეკის ფუნქციების ძალიან წარმატებული ნაკრების გამო.

აღვნიშნავ, რომ ზემოაღნიშნული მხოლოდ ნაწილია იმისა, რისი გაკეთებაც Q-ს შეუძლია, მას აქვს სხვა უნიკალური თვისებებიც. მაგალითად, უკიდურესად მარტივი IPC პროტოკოლი, რომელიც შლის საზღვრებს ცალკეულ Q პროცესებს შორის და საშუალებას გაძლევთ დააკავშიროთ ასობით ეს პროცესი ერთ ქსელში, რომელიც შეიძლება განთავსდეს ათობით სერვერზე მსოფლიოს სხვადასხვა კუთხეში.

წყარო: www.habr.com