실시간 서비스를 예로 들어 Q 및 KDB+ 언어 기능

KDB+ 데이터베이스가 무엇인지, Q 프로그래밍 언어가 무엇인지, 그리고 이들의 강점과 약점이 무엇인지에 대해서는 이전 기사에서 읽을 수 있습니다. 기사 서론에서 간략하게 설명하겠습니다. 이 글에서는 Q에서 수신 데이터 스트림을 처리하고 다양한 집계 함수를 "실시간" 모드로 매분 계산하는 서비스를 구현합니다(즉, 다음 데이터 부분을 계산하기 전에 모든 것을 계산할 시간을 확보합니다). Q의 주요 특징은 단일 객체가 아닌 배열, 배열의 배열, 기타 복잡한 객체를 처리할 수 있는 벡터 언어라는 것입니다. Q와 관련 언어인 K, J, APL은 간결함으로 유명합니다. Java와 같은 익숙한 언어로 여러 화면 분량의 코드를 차지하는 프로그램도 몇 줄로 작성할 수 있는 경우가 많습니다. 이 글에서 바로 이러한 점을 보여드리고자 합니다.

소개

KDB+는 특정 방식(주로 시간 기준)으로 정렬된 대용량 데이터를 위해 설계된 컬럼형 데이터베이스입니다. 주로 은행, 투자 펀드, 보험 회사 등 금융 기관에서 사용됩니다. Q 언어는 KDB+의 내부 언어로, 이 데이터를 효과적으로 다룰 수 있도록 해줍니다. Q 언어의 핵심은 간결성과 효율성을 중시하는 반면, 명확성은 희생됩니다. 벡터 언어는 어떤 경우에도 인지하기 어렵다는 사실과, 레코드의 간결성과 풍부함 덕분에 프로그램의 훨씬 더 많은 부분을 한 화면에서 볼 수 있어 궁극적으로 이해하기가 더 쉽다는 사실에서 그 이유를 찾을 수 있습니다.

이 글에서는 Q로 완전한 프로그램을 구현해 보겠습니다. 직접 체험해 보시는 것도 좋을 것 같습니다. 이를 위해서는 Q 자체가 필요합니다. kx 회사 웹사이트에서 무료 32비트 버전을 다운로드할 수 있습니다. www.kx.com. 관심이 있으시다면 Q라는 책에 대한 참고 정보도 찾으실 수 있습니다. Q For Mortals 이 주제에 대한 다양한 기사가 있습니다.

문제 성명

25밀리초마다 데이터 테이블을 전송하는 소스가 있습니다. KDB+는 주로 금융 분야에서 사용되므로, 이 테이블은 거래 테이블이라고 가정하겠습니다. 이 테이블의 열은 다음과 같습니다. time(밀리초 단위 시간), sym(증권 거래소의 회사 종목) IBM, AAPL,…), 가격(주식 매수 가격), 크기(거래 규모). 25밀리초 간격은 임의로 선택되었으며, 너무 작지도 크지도 않습니다. 이 간격은 데이터가 이미 버퍼링된 상태로 서비스에 도착함을 의미합니다. 현재 부하에 따라 동적 버퍼링을 포함하여 서비스 측에서 버퍼링을 구현하는 것은 쉽지만, 단순화를 위해 고정된 간격으로 중단하겠습니다.

서비스는 sym 열에서 들어오는 각 심볼에 대해 분당 집계 함수 집합(최대 가격, 평균 가격, 합계 크기 등)을 계산해야 합니다. 이는 유용한 정보입니다. 단순화를 위해 모든 함수는 증분식으로 계산될 수 있다고 가정합니다. 즉, 새 값을 얻으려면 이전 값과 들어오는 값, 두 개의 숫자만 알면 됩니다. 예를 들어, 최대값, 평균값, 합계 함수는 이러한 속성을 갖지만, 중간값 함수는 이러한 속성을 갖지 않습니다.

또한 수신 데이터 스트림이 시간순으로 정렬되어 있다고 가정합니다. 이렇게 하면 마지막 분만 처리하면 됩니다. 실제로는 일부 업데이트가 지연될 경우를 대비하여 현재 분과 이전 분만 처리하면 됩니다. 편의상 이 경우는 고려하지 않겠습니다.

집계 함수

필요한 집계 함수는 다음과 같습니다. 서비스 부하를 늘리기 위해 가능한 한 많은 함수를 사용했습니다.

• 높음 – 최대 가격 – 분당 최대 가격.
• 낮음 – 최소 가격 – 분당 최소 가격.
• firstPrice – 첫 번째 가격 – 분당 첫 번째 가격.
• lastPrice – 마지막 가격 – 분당 마지막 가격.
• firstSize – 첫 번째 크기 – 분당 첫 번째 거래 크기.
• lastSize – 마지막 크기 — 분당 마지막 거래 크기입니다.
• numTrades – count i – 분당 거래 수.
• 거래량 – 총 거래 규모 – 분당 거래 규모의 합계.
• pvolume – sum price – 분당 가격의 합계, avgPrice에 필요함.
• 거래액 – 가격 합계 * 규모 – 분당 총 거래량.
• avgPrice – pvolume%numTrades – 분당 평균 가격.
• avgSize – volume%numTrades – 분당 평균 거래 규모.
• vwap – 거래량%매출 – 거래 규모에 따라 가중치를 둔 분당 평균 가격.
• cumVolume – 총 거래량 – 모든 기간 동안 누적된 거래량입니다.

바로 명확하지 않은 점 하나를 살펴보겠습니다. 바로 이러한 열을 처음과 이후 매 분마다 초기화하는 방법입니다. firstPrice 유형의 일부 열은 매번 null 값으로 초기화해야 하며, 그 값은 정의되지 않습니다. volume 유형의 다른 열은 항상 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과 time을 추가했습니다. 이제 initWith는 최종 집계 테이블에서 미리 만들어진 줄이며, 올바른 sym과 time을 설정하는 역할을 합니다. 이를 사용하여 테이블에 새 행을 추가할 수 있습니다.

집계 함수를 생성할 때 aggCols가 필요합니다. Q의 표현식 계산 순서(오른쪽에서 왼쪽)의 특성으로 인해 리스트를 반전해야 합니다. 일부 열이 이전 열에 종속되므로, 계산이 high에서 cumVolume으로 진행되도록 하는 것이 목표입니다.

이전 분에서 새 분으로 복사해야 하는 열은 편의상 sym 열이 추가되었습니다.

rollColumns:`sym`cumVolume;

이제 업데이트 방식에 따라 열을 그룹으로 나눠 보겠습니다. 세 가지 유형이 있습니다.

1. 누적기(거래량, 매출 등) – 유입 가치를 이전 가치에 더해야 합니다.
2. 특정 지점(높음, 낮음, ..)을 사용하면 분의 첫 번째 값은 들어오는 데이터에서 가져오고 나머지는 함수를 사용하여 계산합니다.
3. 나머지는 항상 함수를 사용하여 계산됩니다.

다음 클래스에 대한 변수를 정의해 보겠습니다.

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

계산 순서

집계 테이블을 두 단계에 걸쳐 업데이트합니다. 효율성을 위해 먼저 입력 테이블을 압축하여 각 기호와 분에 대해 하나의 행이 있도록 합니다. 모든 함수가 증분형이고 연관 함수이므로 이 추가 단계에서 결과가 변경되지 않습니다. select를 사용하여 테이블을 압축할 수 있습니다.

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

명확성을 위해, parse 함수를 사용했습니다. 이 함수는 Q 표현식이 포함된 문자열을 eval 함수에 전달될 수 있는 값으로 변환하여 select 함수에 필요한 값으로 사용합니다. 또한, preprocess 함수는 select 함수의 프로젝션(즉, 인수가 부분적으로 지정된 함수)으로 정의되어 있으며, 인수 하나(테이블)가 누락되었습니다. 테이블에 preprocess 함수를 적용하면 압축된 테이블이 생성됩니다.

두 번째 단계는 집계 테이블을 업데이트하는 것입니다. 먼저 알고리즘을 의사코드로 작성해 보겠습니다.

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 함수를 사용하여 행렬을 전치(열 목록을 행 목록으로)해야 합니다. 이 작업은 큰 표에서는 비용이 많이 들기 때문에, 대신 map 함수(아포스트로피처럼 보이는)를 사용하여 각 열에 개별적으로 일반 할당을 적용합니다.

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

함수 투영을 다시 사용합니다. 또한 Q에서 리스트를 생성하는 것도 함수이며, each(map)을 사용하여 리스트의 리스트를 얻을 수 있다는 점에 유의하세요.

계산된 열의 고정된 집합을 피하기 위해 위 표현식을 동적으로 생성해 보겠습니다. 먼저, row 및 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에서 값을 선택하는 select 함수(?[Boolean list;list1;list2])가 있습니다.

// 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 기준으로는 일반적인 할당입니다. 다만 값 목록을 한 번에 할당하는 것뿐입니다. 마지막으로 main 함수를 만들어 보겠습니다.

// ":",/: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밀리초까지 변합니다.

이제 서비스로 데이터를 보내볼 수 있습니다(처음 10시간을 추가해 보겠습니다).

{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밀리초마다 XNUMX줄씩 실행해 보세요.

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억 XNUMX천만 줄)은 매우 방대하기 때문에, 이러한 경우에는 서비스의 복제본을 여러 개(심지어 수십 개) 실행하여 각각 문자의 일부만 처리하는 것이 일반적입니다. 그럼에도 불구하고, 데이터 저장에 주로 초점을 맞춘 해석 언어로서는 인상적인 결과입니다.

각 업데이트 크기에 따라 시간이 비선형적으로 증가하는 이유가 궁금할 수 있습니다. 그 이유는 축소 함수가 실제로 updateAgg보다 훨씬 효율적으로 작동하는 C 함수이기 때문입니다. 특정 업데이트 크기(약 10.000)에서 시작하여 updateAgg는 상한에 도달하고 그 후 실행 시간은 업데이트 크기에 따라 달라지지 않습니다. 서비스가 이러한 볼륨의 데이터를 소화할 수 있는 것은 예비 Q 단계 때문입니다. 이는 빅데이터로 작업할 때 올바른 알고리즘을 선택하는 것이 얼마나 중요한지 강조합니다. 또 다른 요점은 메모리에 데이터를 올바르게 저장하는 것입니다. 데이터가 열 형식으로 저장되지 않았거나 시간순으로 정렬되지 않았다면 TLB 캐시 미스(프로세서 주소 캐시에 메모리 페이지 주소가 없는 상태)와 같은 문제에 직면하게 될 것입니다. 실패 시 주소 검색에 약 30배 더 오랜 시간이 걸리고 데이터가 분산된 경우 서비스 속도가 여러 배 느려질 수 있습니다.

결론

이 글에서는 KDB+와 Q 데이터베이스가 대용량 데이터를 저장하고 select를 통해 쉽게 접근할 수 있을 뿐만 아니라, 단일 Q 프로세스에서도 수억 개의 행/기가바이트 데이터를 처리할 수 있는 데이터 처리 서비스를 구축하는 데에도 적합하다는 것을 보여주었습니다. Q 언어 자체는 벡터 기반 특성, 내장 SQL 언어 인터프리터, 그리고 매우 유용한 라이브러리 함수 세트를 통해 매우 간단하고 효율적인 데이터 처리 알고리즘 구현을 가능하게 합니다.

위의 내용은 Q의 기능 중 일부에 불과하며, Q는 다른 고유한 기능들을 가지고 있습니다. 예를 들어, 개별 Q 프로세스 간의 경계를 없애고 수백 개의 프로세스를 단일 네트워크로 통합할 수 있도록 하는 매우 간단한 IPC 프로토콜이 있습니다. 이 네트워크는 전 세계 수십 개의 서버에 분산 배치될 수 있습니다.

출처 : habr.com