תכונות של שפת Q ו-KDB+ תוך שימוש בדוגמה של שירות בזמן אמת

אתה יכול לקרוא על מה זה בסיס KDB+, שפת התכנות Q, מה החוזקות והחולשות שלהם בקודם שלי статье ובקצרה בהקדמה. במאמר, ניישם שירות ב-Q שיעבד את זרם הנתונים הנכנס ויחשב פונקציות צבירה שונות בכל דקה במצב "זמן אמת" (כלומר, יהיה לו זמן לחשב הכל לפני החלק הבא של הנתונים). המאפיין העיקרי של Q הוא שמדובר בשפה וקטורית המאפשרת לך לפעול לא עם אובייקטים בודדים, אלא עם המערכים שלהם, מערכי המערכים ואובייקטים מורכבים אחרים. שפות כמו Q וקרוביו K, J, APL מפורסמות בקיצורן. לעתים קרובות, תוכנה שתופסת מספר מסכים של קוד בשפה מוכרת כמו Java יכולה להיכתב עליהם בכמה שורות. זה מה שאני רוצה להדגים במאמר זה.

מבוא

KDB+ הוא מסד נתונים עמודי המתמקד בכמויות גדולות מאוד של נתונים, מסודרים בצורה ספציפית (בעיקר לפי זמן). הוא משמש בעיקר במוסדות פיננסיים - בנקים, קרנות השקעות, חברות ביטוח. שפת Q היא השפה הפנימית של KDB+ המאפשרת לך לעבוד ביעילות עם הנתונים הללו. אידיאולוגיית ה-Q היא קיצור ויעילות, בעוד שהבהירות מוקרבת. זה מוצדק בעובדה שהשפה הווקטורית תהיה קשה להבנה בכל מקרה, וקיצור ההקלטה ועושרה מאפשרים לראות חלק גדול בהרבה מהתוכנית על מסך אחד, מה שבסופו של דבר מקל על ההבנה.

במאמר זה אנו מיישמים תוכנית מלאה ב-Q ואולי תרצה לנסות אותה. כדי לעשות זאת, תזדקק ל-Q בפועל. אתה יכול להוריד את גרסת 32 סיביות בחינם באתר החברה kx - www.kx.com. שם, אם אתה מעוניין, תמצא מידע עיון על Q, הספר ש לבני תמותה ומאמרים שונים בנושא זה.

הצהרת הבעיה

יש מקור ששולח טבלה עם נתונים כל 25 מילישניות. מכיוון ש-KDB+ משמש בעיקר בתחום הפיננסים, נניח שמדובר בטבלת עסקאות (עסקאות), הכוללות את העמודות הבאות: זמן (זמן באלפיות השנייה), sym (ייעוד חברה בבורסה - יבמ, AAPL,...), מחיר (המחיר שבו נרכשו המניות), גודל (גודל העסקה). המרווח של 25 מילישניות הוא שרירותי, לא קטן מדי ולא ארוך מדי. הנוכחות שלו פירושה שהנתונים מגיעים לשירות כבר מאוגרים. יהיה קל ליישם חציצה בצד השירות, כולל חציצה דינמית בהתאם לעומס הנוכחי, אך לשם הפשטות, נתמקד במרווח קבוע.

השירות חייב לספור כל דקה עבור כל סמל נכנס מעמודת sym קבוצה של פונקציות צבירה - מחיר מקסימלי, מחיר ממוצע, גודל סכום וכו'. מידע שימושי. לשם הפשטות, נניח שניתן לחשב את כל הפונקציות באופן אינקרמנטלי, כלומר. כדי לקבל ערך חדש, מספיק לדעת שני מספרים - הישן והערכים הנכנסים. לדוגמה, לפונקציות מקסימום, ממוצע, סכום יש את המאפיין הזה, אבל לפונקציה החציונית אין.

כמו כן, נניח שזרם הנתונים הנכנס מוסדר בזמן. זה ייתן לנו את ההזדמנות לעבוד רק עם הרגע האחרון. בפועל, מספיק להיות מסוגל לעבוד עם הדקות הנוכחיות והקודמות למקרה שחלק מהעדכונים יאחרו. לשם הפשטות, לא נשקול את המקרה הזה.

פונקציות צבירה

פונקציות הצבירה הנדרשות מפורטות להלן. לקחתי כמה שיותר מהם כדי להגדיל את העומס על השירות:

• מחיר גבוה – מקסימום מחיר – מחיר מקסימלי לדקה.
• נמוך – מחיר מינימלי – מחיר מינימום לדקה.
• firstPrice – מחיר ראשון – מחיר ראשון לדקה.
• lastPrice – מחיר אחרון – מחיר אחרון לדקה.
• firstSize - מידה ראשונה - גודל סחר ראשון לדקה.
• lastSize - גודל אחרון - גודל טרייד אחרון תוך דקה.
• numTrades – count i – מספר העסקאות בדקה.
• נפח – גודל סכום – סכום גדלי מסחר לדקה.
• pvolume – סכום מחיר – סכום מחירים לדקה, נדרש עבור avgPrice.
• – סכום מחזור מחיר*גודל – היקף עסקאות כולל לדקה.
• avgPrice – pvolume%numTrades – מחיר ממוצע לדקה.
• avgSize – volume%numTrades – גודל מסחר ממוצע לדקה.
• vwap – turnover%volume – מחיר ממוצע לדקה משוקלל לפי גודל העסקה.
• cumVolume – נפח סכום – גודל עסקאות מצטבר לאורך כל הזמן.

מיד נדון בנקודה אחת לא ברורה - כיצד לאתחל את העמודות הללו בפעם הראשונה ולכל דקה שלאחר מכן. יש לאתחל כמה עמודות מסוג firstPrice ל-null בכל פעם; הערך שלהן אינו מוגדר. סוגי נפחים אחרים חייבים להיות מוגדרים תמיד ל-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 (מימין לשמאל). המטרה היא להבטיח שהחישוב יעבור מגבוה ל-cumVolume, מכיוון שחלק מהעמודות תלויות בעמודות הקודמות.

עמודות שצריך להעתיק לדקה חדשה מהקודמת, עמודת הסמל מתווספת מטעמי נוחות:

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]

לא אתאר בפירוט את הפורמט של הטיעונים; במקרה שלנו, רק ביטויים לפי ובחר יהיו לא טריוויאליים והם צריכים להיות מילונים של עמודות הצורה!ביטויים. לפיכך, ניתן להגדיר את פונקציית הכיווץ באופן הבא:

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

לצורך הבהירות השתמשתי בפונקציית parse, שהופכת מחרוזת עם ביטוי Q לערך שניתן להעביר לפונקציה eval ואשר נדרשת ב-function select. שימו לב גם ש-preprocess מוגדר כהשלכה (כלומר, פונקציה עם ארגומנטים מוגדרים חלקית) של פונקציית הבחירה, ארגומנט אחד (הטבלה) חסר. אם נחיל קדם-process על טבלה, נקבל טבלה דחוסה.

השלב השני הוא עדכון הטבלה המצטברת. בוא נכתוב תחילה את האלגוריתם בפסאודוקוד:

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, יצירת רשימה היא גם פונקציה ונוכל לקרוא לה באמצעות הפונקציה each(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");

חלק מהעמודות הן מיוחדות; אין לחשב את הערך הראשון שלהן על ידי הפונקציה. אנו יכולים לקבוע שהוא הראשון לפי העמודה row[`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];

כאן קראתי מטלה כללית עם הפונקציה שלי (ביטוי בסוגרים מתולתלים). הוא מקבל את הערך הנוכחי (הארגומנט הראשון) וארגומנט נוסף, אותו אני מעביר בפרמטר הרביעי.

בואו נוסיף רמקולים סוללה בנפרד, מכיוון שהפונקציה זהה עבורם:

// 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 סדר ההערכה הוא מימין לשמאל.

כעת יש לנו שתי פונקציות עיקריות הדרושות לחישובים, אנחנו רק צריכים להוסיף מעט תשתית והשירות מוכן.

שלבים אחרונים

יש לנו פונקציות Preprocess ו- UpdateAgg שעושות את כל העבודה. אבל עדיין יש צורך להבטיח את המעבר הנכון דרך דקות ולחשב אינדקסים לצבירה. קודם כל, בואו נגדיר את פונקציית 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 מתאימים לא רק לאחסון נתונים גדולים וגישה אליהם בקלות באמצעות select, אלא גם ליצירת שירותי עיבוד נתונים המסוגלים לעכל מאות מיליוני שורות/גיגה-בייט של נתונים אפילו תהליך Q אחד בודד. שפת ה-Q עצמה מאפשרת יישום תמציתי ויעיל ביותר של אלגוריתמים הקשורים לעיבוד נתונים בשל האופי הווקטורי שלה, מתורגמן ניבים מובנה של SQL ומערכת מוצלחת מאוד של פונקציות ספרייה.

אציין שהאמור לעיל הוא רק חלק ממה ש-Q יכול לעשות, יש לו גם תכונות ייחודיות אחרות. למשל, פרוטוקול IPC פשוט במיוחד שמוחק את הגבול בין תהליכי Q בודדים ומאפשר לשלב מאות תהליכים אלו לרשת אחת, שיכולה להתמקם על עשרות שרתים במקומות שונים בעולם.

מקור: www.habr.com