একটি রিয়েল-টাইম পরিষেবার উদাহরণ ব্যবহার করে Q এবং KDB+ ভাষার বৈশিষ্ট্য

কেডিবি+ বেস, কিউ প্রোগ্রামিং ভাষা কী, তাদের শক্তি এবং দুর্বলতাগুলি কী তা আমার আগের লেখায় আপনি পড়তে পারেন প্রবন্ধ এবং সংক্ষিপ্তভাবে ভূমিকা. নিবন্ধে, আমরা Q-তে একটি পরিষেবা প্রয়োগ করব যা ইনকামিং ডেটা স্ট্রিম প্রক্রিয়া করবে এবং "রিয়েল টাইম" মোডে প্রতি মিনিটে বিভিন্ন একত্রীকরণ ফাংশন গণনা করবে (অর্থাৎ, ডেটার পরবর্তী অংশের আগে সবকিছু গণনা করার সময় থাকবে)। Q-এর প্রধান বৈশিষ্ট্য হল এটি একটি ভেক্টর ভাষা যা আপনাকে একক বস্তুর সাথে নয়, তাদের অ্যারে, অ্যারে এবং অন্যান্য জটিল বস্তুর সাথে কাজ করতে দেয়। Q এবং এর আত্মীয় K, ​​J, APL এর মতো ভাষাগুলি তাদের সংক্ষিপ্ততার জন্য বিখ্যাত। প্রায়শই, একটি প্রোগ্রাম যা জাভার মতো একটি পরিচিত ভাষায় কোডের বেশ কয়েকটি স্ক্রিন নেয় সেগুলিতে কয়েকটি লাইনে লেখা যেতে পারে। এই আমি এই নিবন্ধে প্রদর্শন করতে চান কি.

ভূমিকা

KDB+ হল একটি কলামার ডাটাবেস যা খুব বেশি পরিমাণে ডেটার উপর দৃষ্টি নিবদ্ধ করে, একটি নির্দিষ্ট উপায়ে (প্রাথমিকভাবে সময় অনুসারে) অর্ডার করা হয়। এটি প্রাথমিকভাবে আর্থিক প্রতিষ্ঠানগুলিতে ব্যবহৃত হয় - ব্যাংক, বিনিয়োগ তহবিল, বীমা কোম্পানি। Q ভাষা হল KDB+ এর অভ্যন্তরীণ ভাষা যা আপনাকে এই ডেটার সাথে কার্যকরভাবে কাজ করতে দেয়। Q মতাদর্শ হল সংক্ষিপ্ততা এবং দক্ষতা, যখন স্বচ্ছতা বলি দেওয়া হয়। এটি এই সত্যের দ্বারা ন্যায্য যে ভেক্টর ভাষাটি যে কোনও ক্ষেত্রে বোঝা কঠিন হবে এবং রেকর্ডিংয়ের সংক্ষিপ্ততা এবং সমৃদ্ধি আপনাকে একটি স্ক্রিনে প্রোগ্রামের অনেক বড় অংশ দেখতে দেয়, যা শেষ পর্যন্ত বোঝা সহজ করে তোলে।

এই নিবন্ধে আমরা Q-তে একটি পূর্ণাঙ্গ প্রোগ্রাম বাস্তবায়ন করেছি এবং আপনি এটি চেষ্টা করে দেখতে চাইতে পারেন। এটি করার জন্য, আপনার প্রকৃত Q প্রয়োজন হবে। আপনি kx কোম্পানির ওয়েবসাইটে বিনামূল্যে 32-বিট সংস্করণ ডাউনলোড করতে পারেন - www.kx.com. সেখানে, আপনি আগ্রহী হলে, আপনি Q, বইয়ের রেফারেন্স তথ্য পাবেন Q নশ্বরদের জন্য এবং এই বিষয়ে বিভিন্ন নিবন্ধ।

সমস্যা গঠন

একটি উৎস রয়েছে যা প্রতি 25 মিলিসেকেন্ডে ডেটা সহ একটি টেবিল পাঠায়। যেহেতু KDB+ প্রাথমিকভাবে ফাইন্যান্সে ব্যবহৃত হয়, তাই আমরা ধরে নেব যে এটি লেনদেনের একটি সারণী (বাণিজ্য), যার নিম্নলিখিত কলাম রয়েছে: সময় (মিলিসেকেন্ডে সময়), সিম (স্টক এক্সচেঞ্জে কোম্পানির পদবী - আইবিএম, AAPL,…), মূল্য (যে দামে শেয়ার কেনা হয়েছিল), আকার (লেনদেনের আকার)। 25 মিলিসেকেন্ডের ব্যবধানটি নির্বিচারে, খুব ছোট নয় এবং খুব দীর্ঘ নয়৷ এর উপস্থিতির অর্থ হল ডেটা ইতিমধ্যেই বাফার করা পরিষেবাতে আসে৷ বর্তমান লোডের উপর নির্ভর করে গতিশীল বাফারিং সহ পরিষেবার দিকে বাফারিং বাস্তবায়ন করা সহজ হবে, কিন্তু সরলতার জন্য, আমরা একটি নির্দিষ্ট ব্যবধানে ফোকাস করব।

পরিষেবাটিকে অবশ্যই সিম কলাম থেকে প্রতিটি আগত প্রতীকের জন্য প্রতি মিনিটে গণনা করতে হবে সমষ্টিগত ফাংশনের একটি সেট - সর্বোচ্চ মূল্য, গড় মূল্য, যোগফলের আকার ইত্যাদি। দরকারী তথ্য। সরলতার জন্য, আমরা ধরে নেব যে সমস্ত ফাংশন ক্রমবর্ধমানভাবে গণনা করা যেতে পারে, যেমন একটি নতুন মান পেতে, দুটি সংখ্যা জানা যথেষ্ট - পুরানো এবং আগত মান। উদাহরণস্বরূপ, ফাংশন সর্বাধিক, গড়, যোগফল এই বৈশিষ্ট্য আছে, কিন্তু মধ্য ফাংশন নেই.

আমরা এটাও ধরে নেব যে ইনকামিং ডেটা স্ট্রীম নির্ধারিত সময়ের জন্য। এটি আমাদের শুধুমাত্র শেষ মুহূর্তে কাজ করার সুযোগ দেবে। অনুশীলনে, কিছু আপডেট দেরী হলে বর্তমান এবং পূর্ববর্তী মিনিটের সাথে কাজ করতে সক্ষম হওয়া যথেষ্ট। সরলতার জন্য, আমরা এই ক্ষেত্রে বিবেচনা করব না।

একত্রীকরণ ফাংশন

প্রয়োজনীয় সমষ্টি ফাংশন নীচে তালিকাভুক্ত করা হয়. আমি পরিষেবাতে লোড বাড়ানোর জন্য তাদের মধ্যে যতটা সম্ভব নিয়েছি:

• উচ্চ – সর্বোচ্চ মূল্য – প্রতি মিনিটে সর্বোচ্চ মূল্য।
• কম - ন্যূনতম মূল্য - প্রতি মিনিটে সর্বনিম্ন মূল্য৷
• প্রথম মূল্য - প্রথম মূল্য - প্রতি মিনিটে প্রথম মূল্য৷
• শেষ মূল্য - শেষ মূল্য - প্রতি মিনিটে শেষ মূল্য।
• প্রথম আকার - প্রথম আকার - প্রতি মিনিটে প্রথম বাণিজ্য আকার।
• লাস্ট সাইজ - শেষ সাইজ - এক মিনিটে শেষ ট্রেড সাইজ।
• numTrades – কাউন্ট i – প্রতি মিনিটে ট্রেডের সংখ্যা।
• ভলিউম – সমষ্টির আকার – প্রতি মিনিটে ট্রেড সাইজের সমষ্টি।
• pvolume – সমষ্টি মূল্য – প্রতি মিনিটে দামের যোগফল, avgPrice এর জন্য প্রয়োজন।
• - যোগফল টার্নওভার মূল্য * আকার - প্রতি মিনিটে লেনদেনের মোট পরিমাণ।
• avgPrice – pvolume%numTrades – প্রতি মিনিটে গড় মূল্য।
• avgSize – ভলিউম%numTrades – প্রতি মিনিটে গড় ট্রেড সাইজ।
• vwap - টার্নওভার% ভলিউম - লেনদেনের আকার অনুসারে প্রতি মিনিটে গড় মূল্য।
• কম ভলিউম - সমষ্টি ভলিউম - পুরো সময় ধরে লেনদেনের সঞ্চিত আকার।

আসুন অবিলম্বে একটি অ-স্পষ্ট বিষয় নিয়ে আলোচনা করা যাক - কীভাবে এই কলামগুলি প্রথমবার এবং পরবর্তী প্রতিটি মিনিটের জন্য শুরু করা যায়। ফার্স্টপ্রাইস টাইপের কিছু কলাম প্রতিবার শূন্য করতে আরম্ভ করতে হবে; তাদের মান অনির্ধারিত। অন্যান্য ভলিউম প্রকারগুলি সর্বদা 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 চূড়ান্ত সমষ্টি সারণী থেকে একটি রেডিমেড লাইন, যেখানে এটি সঠিক sym এবং সময় সেট করতে থাকে। আপনি একটি টেবিলে নতুন সারি যোগ করতে এটি ব্যবহার করতে পারেন।

একটি সমষ্টি ফাংশন তৈরি করার সময় আমাদের aggCols প্রয়োজন হবে। তালিকাটি অবশ্যই উল্টাতে হবে কারণ Q-তে অভিব্যক্তি মূল্যায়ন করা হয় (ডান থেকে বামে)। লক্ষ্য হল গণনা উচ্চ থেকে কম ভলিউমে যায় তা নিশ্চিত করা, যেহেতু কিছু কলাম পূর্ববর্তীগুলির উপর নির্ভর করে।

যে কলামগুলিকে আগেরটি থেকে একটি নতুন মিনিটে অনুলিপি করতে হবে, সুবিধার জন্য সিম কলামটি যুক্ত করা হয়েছে:

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-তে, নির্বাচন একটি ফাংশন হিসাবে প্রয়োগ করা হয় যেখানে আপনি গতিশীলভাবে তৈরি আর্গুমেন্টগুলি প্রতিস্থাপন করতে পারেন:

?[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 এক্সপ্রেশন সহ একটি স্ট্রিংকে একটি মানতে পরিণত করে যা 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;…)];

দুর্ভাগ্যবশত, একটি টেবিলে বরাদ্দ করার জন্য আপনাকে সারিগুলির একটি তালিকা প্রয়োজন, কলাম নয়, এবং আপনাকে ফ্লিপ ফাংশন ব্যবহার করে ম্যাট্রিক্স (সারির তালিকায় কলামের তালিকা) স্থানান্তর করতে হবে। এটি একটি বড় টেবিলের জন্য ব্যয়বহুল, তাই এর পরিবর্তে আমরা ম্যাপ ফাংশন ব্যবহার করে প্রতিটি কলামে আলাদাভাবে একটি সাধারণ অ্যাসাইনমেন্ট প্রয়োগ করি (যা একটি অ্যাপোস্ট্রফির মতো দেখায়):

.[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-তে মূল্যায়নের ক্রম ডান থেকে বামে থাকে।

এখন আমাদের কাছে গণনার জন্য প্রয়োজনীয় দুটি প্রধান ফাংশন রয়েছে, আমাদের শুধু একটু পরিকাঠামো যোগ করতে হবে এবং পরিষেবা প্রস্তুত।

চূড়ান্ত পদক্ষেপ

আমাদের প্রিপ্রসেস এবং আপডেটএগ ফাংশন রয়েছে যা সমস্ত কাজ করে। কিন্তু মিনিটের মাধ্যমে সঠিক স্থানান্তর নিশ্চিত করা এবং সমষ্টির জন্য সূচক গণনা করা এখনও প্রয়োজন। প্রথমত, 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 মিলিয়ন সারি) অত্যন্ত বড়; এই ধরনের ক্ষেত্রে, পরিষেবার বেশ কয়েকটি ক্লোন (বা এমনকি কয়েক ডজন ক্লোন) চালু করা সাধারণ, যার প্রতিটি অক্ষরের শুধুমাত্র অংশ প্রক্রিয়া করে। তবুও, ফলাফলটি একটি ব্যাখ্যা করা ভাষার জন্য চিত্তাকর্ষক যা প্রাথমিকভাবে ডেটা স্টোরেজের উপর ফোকাস করে।

প্রশ্ন উঠতে পারে কেন সময় প্রতিটি আপডেটের আকারের সাথে অ-রৈখিকভাবে বৃদ্ধি পায়। কারণটি হল সঙ্কুচিত ফাংশনটি আসলে একটি সি ফাংশন, যা আপডেটএগের চেয়ে অনেক বেশি কার্যকর। একটি নির্দিষ্ট আপডেটের আকার (প্রায় 10.000) থেকে শুরু করে, updateAgg এর সিলিংয়ে পৌঁছে এবং তারপরে এটির সম্পাদনের সময় আপডেটের আকারের উপর নির্ভর করে না। এটি প্রাথমিক পদক্ষেপ Q এর কারণে যে পরিষেবাটি এত পরিমাণ ডেটা হজম করতে সক্ষম। এটি হাইলাইট করে যে বড় ডেটা নিয়ে কাজ করার সময় সঠিক অ্যালগরিদম বেছে নেওয়া কতটা গুরুত্বপূর্ণ। আরেকটি পয়েন্ট হল মেমরিতে ডেটার সঠিক স্টোরেজ। যদি ডেটা কলামারলি সংরক্ষণ করা না হয় বা সময় অনুসারে অর্ডার না করা হয়, তাহলে আমরা TLB ক্যাশে মিস - প্রসেসর অ্যাড্রেস ক্যাশে মেমরি পৃষ্ঠার ঠিকানার অনুপস্থিতির মতো একটি জিনিসের সাথে পরিচিত হব। একটি ঠিকানা অনুসন্ধান করতে ব্যর্থ হলে প্রায় 30 গুণ বেশি সময় লাগে, এবং যদি ডেটা ছড়িয়ে পড়ে, এটি পরিষেবাটি কয়েকবার ধীর করে দিতে পারে৷

উপসংহার

এই নিবন্ধে, আমি দেখিয়েছি যে KDB+ এবং Q ডাটাবেস শুধুমাত্র বড় ডেটা সঞ্চয় করার জন্য এবং নির্বাচনের মাধ্যমে সহজে অ্যাক্সেস করার জন্যই উপযুক্ত নয়, বরং ডেটা প্রসেসিং পরিষেবা তৈরি করার জন্যও উপযুক্ত যা কয়েক মিলিয়ন সারি/গিগাবাইট ডেটা হজম করতে সক্ষম। একটি একক Q প্রক্রিয়া। Q ভাষা নিজেই এর ভেক্টর প্রকৃতি, অন্তর্নির্মিত SQL উপভাষা দোভাষী এবং লাইব্রেরি ফাংশনগুলির একটি খুব সফল সেটের কারণে ডেটা প্রক্রিয়াকরণের সাথে সম্পর্কিত অ্যালগরিদমগুলির অত্যন্ত সংক্ষিপ্ত এবং দক্ষ বাস্তবায়নের অনুমতি দেয়।

আমি লক্ষ্য করব যে উপরেরটি Q এর ক্ষমতার অংশ মাত্র; এটির অন্যান্য অনন্য বৈশিষ্ট্যও রয়েছে। উদাহরণস্বরূপ, একটি অত্যন্ত সাধারণ IPC প্রোটোকল যা পৃথক Q প্রক্রিয়াগুলির মধ্যে সীমানা মুছে দেয় এবং আপনাকে এই শত শত প্রক্রিয়াগুলিকে একটি একক নেটওয়ার্কে একত্রিত করতে দেয়, যা বিশ্বের বিভিন্ন অংশে কয়েক ডজন সার্ভারে অবস্থিত হতে পারে।

উত্স: www.habr.com