Apa itu KDB +, bahasa pemrograman Q, apa kekuatan dan kelemahannya, dapat ditemukan di
artikel saya sebelumnya dan secara singkat dalam pengantar. Dalam artikel tersebut, kami menerapkan layanan pada Q yang akan memproses aliran data yang masuk dan menghitung per menit berbagai fungsi agregasi dalam mode "waktu nyata" (yaitu, ia akan berhasil menghitung semuanya sampai bagian data berikutnya). Fitur utama Q adalah bahwa ia adalah bahasa vektor yang memungkinkan Anda untuk beroperasi bukan dengan objek tunggal, tetapi dengan array, array array, dan objek kompleks lainnya. Bahasa seperti Q dan K, J, APL yang terkait terkenal dengan singkatnya. Seringkali sebuah program yang membentang beberapa layar kode dalam bahasa yang dikenal seperti Java dapat ditulis pada mereka dalam beberapa baris. Inilah yang ingin saya tunjukkan dalam artikel ini.
Pendahuluan
KDB + adalah basis data kolom yang berfokus pada volume data yang sangat besar, diurutkan dengan cara tertentu (terutama berdasarkan waktu). Pertama-tama digunakan dalam organisasi keuangan - bank, dana investasi, perusahaan asuransi. Bahasa Q adalah bahasa internal KDB + yang memungkinkan Anda untuk bekerja secara efektif dengan data ini. Ideologi Q adalah singkatnya dan efisiensi, sementara kejelasan dikorbankan. Ini dibenarkan oleh fakta bahwa dalam hal apa pun bahasa vektor akan sulit dipahami, dan singkatnya dan banyaknya rekaman memungkinkan Anda melihat lebih banyak program pada satu layar, yang pada akhirnya memudahkan pemahamannya.
Pada artikel ini, kami menerapkan program Q yang lengkap dan Anda mungkin ingin mencobanya. Untuk melakukan ini, Anda memerlukan Q yang sebenarnya. Anda dapat mengunduh versi 32-bit gratis di situs web perusahaan kx -
www.kx.com . Di tempat yang sama, jika Anda tertarik, Anda akan menemukan informasi referensi tentang Q, buku
Q For Mortals dan berbagai artikel tentang topik ini.
Pernyataan masalah
Ada sumber yang mengirim tabel data setiap 25 milidetik. Karena KDB + terutama digunakan dalam keuangan, kami menganggap bahwa ini adalah tabel perdagangan di mana terdapat kolom berikut: waktu (waktu dalam milidetik), sym (nama perusahaan pada pertukaran -
IBM ,
AAPL , ...), harga (harga) dimana saham dibeli), ukuran (ukuran transaksi). Interval 25 milidetik dipilih secara sewenang-wenang, tidak terlalu kecil dan tidak terlalu besar. Kehadirannya berarti bahwa data yang tiba di layanan sudah buffer. Akan mudah untuk menerapkan buffering di sisi layanan, termasuk buffering dinamis, tergantung pada beban saat ini, tetapi untuk kesederhanaan kami tinggal pada interval yang tetap.
Layanan harus menghitung setiap menit untuk setiap karakter yang masuk dari kolom sym satu set fungsi agregasi - harga maks, harga rata-rata, ukuran jumlah, dll. informasi yang berguna. Untuk kesederhanaan, kami mengasumsikan bahwa semua fungsi dapat dihitung secara bertahap, mis. untuk mendapatkan nilai baru, cukup mengetahui dua angka - yang lama dan nilai yang masuk. Misalnya, fungsi maks, rata-rata, jumlah memiliki properti ini, tetapi fungsi median tidak.
Kami juga menganggap bahwa aliran data yang masuk dipesan berdasarkan waktu. Ini akan memberi kita kesempatan untuk bekerja hanya pada menit terakhir. Dalam praktiknya, cukup untuk dapat bekerja dengan menit saat ini dan sebelumnya jika ada pembaruan yang terlambat. Untuk mempermudah, kami tidak akan mempertimbangkan kasus ini.
Fungsi Agregat
Di bawah ini adalah fungsi agregat yang diperlukan. Saya mengambilnya sebanyak mungkin untuk menambah beban pada layanan:
- harga tinggi - maksimum - harga maksimum per menit.
- low - min price - harga minimum per menit.
- firstHarga - harga pertama - harga pertama per menit.
- lastPrice - harga terakhir - harga terakhir per menit.
- Ukuran pertama - ukuran pertama - ukuran kesepakatan pertama dalam satu menit.
- lastSize - ukuran terakhir - ukuran kesepakatan terakhir dalam satu menit.
- numTrades - hitung i - jumlah transaksi per menit.
- volume - jumlah ukuran - jumlah ukuran transaksi per menit.
- pvolume - harga jumlah - jumlah harga per menit, diperlukan untuk rata-rata Harga.
- omset - harga jumlah * ukuran - total volume transaksi per menit.
- avgPrice - pvolume% numTades - harga rata-rata per menit.
- avgSize - volume% numTrades - ukuran kesepakatan rata-rata per menit.
- vwap - omset% volume - harga rata-rata per menit yang dihitung berdasarkan ukuran transaksi.
- cumVolume - jumlah volume - akumulasi ukuran transaksi sepanjang waktu.
Segera bahas satu hal yang tidak jelas - bagaimana menginisialisasi kolom ini untuk pertama kali dan setiap menit berikutnya. Beberapa kolom tipe firstPrice perlu diinisialisasi dengan null setiap kali, nilainya tidak didefinisikan. Jenis volume lainnya harus selalu disetel ke 0. Masih ada kolom yang memerlukan pendekatan gabungan - misalnya, cumVolume harus disalin dari menit sebelumnya, dan untuk set pertama ke 0. Kami akan mengatur semua parameter ini menggunakan kamus tipe data (analog dari catatan):
Saya menambahkan sym dan waktu ke kamus untuk kenyamanan, sekarang initWith adalah garis jadi dari tabel agregat akhir, di mana ia tetap untuk mengatur sym dan waktu yang benar. Anda bisa menggunakannya untuk menambahkan baris baru ke tabel.
aggCols yang kita butuhkan saat membuat fungsi agregat. Daftar harus terbalik karena kekhasan urutan di mana ekspresi dihitung dalam Q (dari kanan ke kiri). Tujuannya adalah untuk memberikan perhitungan dalam arah dari tinggi ke cumVolume, karena beberapa kolom bergantung pada yang sebelumnya.
Kolom yang akan disalin ke menit baru dari yang sebelumnya, kolom sym ditambahkan untuk kenyamanan:
rollColumns:`sym`cumVolume;
Sekarang kami membagi kolom ke dalam kelompok sesuai dengan bagaimana mereka harus diperbarui. Tiga jenis dapat dibedakan:
- Baterai (volume, turnover, ..) - kita harus menambahkan nilai input ke yang sebelumnya.
- Dengan titik khusus (tinggi, rendah, ..) - nilai pertama dalam satu menit diambil dari data input, sisanya dihitung menggunakan fungsi.
- Sisanya. Selalu dihitung menggunakan fungsi.
Tentukan variabel untuk kelas-kelas ini:
accumulatorCols:`numTrades`volume`pvolume`turnover; specialCols:`high`low`firstPrice`firstSize;
Pesanan perhitungan
Kami akan memperbarui tabel teragregasi dalam dua tahap. Untuk efisiensi, pertama-tama kita akan mengecilkan tabel yang masuk sehingga ada satu baris tersisa untuk setiap karakter dan menit. Fakta bahwa semua fungsi kita bersifat inkremental dan asosiatif menjamin kita bahwa hasil dari langkah tambahan ini tidak akan berubah. Anda bisa memeras tabel menggunakan pilih:
select high:max price, low:min price … by sym,time.minute from table
Metode ini memiliki minus - set kolom yang dihitung sudah ditentukan sebelumnya. Untungnya, di Q, pemilihan juga diterapkan sebagai fungsi di mana Anda dapat mengganti argumen yang dibuat secara dinamis:
?[table;whereClause;byClause;selectClause]
Saya tidak akan menjelaskan secara terperinci format argumen, dalam kasus kami hanya dengan dan pilih ekspresi yang tidak sepele dan mereka harus kamus dari kolom formulir! Ekspresi. Dengan demikian, fungsi pembatas dapat didefinisikan sebagai berikut:
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");
Untuk kejelasan, saya menggunakan fungsi parse, yang mengubah string dengan ekspresi Q menjadi nilai yang dapat diteruskan ke fungsi eval dan yang diperlukan dalam fungsi pilih. Perhatikan juga bahwa preprocess didefinisikan sebagai proyeksi (mis., Fungsi dengan argumen yang didefinisikan sebagian) dari fungsi yang dipilih, satu argumen (tabel) tidak ada. Jika kami menerapkan preprocess ke sebuah tabel, kami mendapatkan tabel yang menyusut.
Tahap kedua adalah memperbarui tabel agregat. Pertama, kita menulis algoritma di pseudocode:
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]; …
Dalam Q, alih-alih loop, biasanya menggunakan peta / mengurangi fungsi. Tetapi karena Q adalah bahasa vektor dan semua operasi, kita dapat dengan aman menerapkan semua simbol sekaligus, kemudian sebagai perkiraan pertama yang dapat kita lakukan tanpa siklus sama sekali, melakukan operasi dengan semua simbol sekaligus:
idx:calcIdx inputTable; row:aggTable idx; aggTable[idx;`high]: row[`high] | inputTable`high; aggTable[idx;`volume]: row[`volume] + inputTable`volume; …
Tapi kita bisa melangkah lebih jauh, di Q ada operator yang unik dan sangat kuat - operator penugasan umum. Ini memungkinkan Anda untuk mengubah set nilai dalam struktur data yang kompleks menggunakan daftar indeks, fungsi, dan argumen. Dalam kasus kami, tampilannya seperti ini:
idx:calcIdx inputTable; rows:aggTable idx;
Sayangnya, untuk menetapkan ke tabel Anda perlu daftar baris, bukan kolom, dan Anda harus mengubah matriks (daftar kolom menjadi daftar baris) menggunakan fungsi flip. Untuk tabel besar, ini tidak perlu, jadi alih-alih kami menerapkan penetapan umum untuk setiap kolom secara terpisah, menggunakan fungsi peta (yang terlihat seperti tanda kutip):
.[aggTable;;:;]'[(idx;)each aggCols; (row[`high] | inputTable`high;row[`volume] + inputTable`volume;…)];
Kami kembali menggunakan fungsi proyeksi. Perhatikan juga bahwa di Q, membuat daftar juga merupakan fungsi dan kita dapat menyebutnya menggunakan fungsi masing-masing (peta) untuk mendapatkan daftar daftar.
Agar kumpulan kolom yang dihitung tidak tetap, buat ekspresi di atas secara dinamis. Pertama, kami mendefinisikan fungsi untuk menghitung setiap kolom, menggunakan variabel baris dan inp untuk referensi yang dikumpulkan dan memasukkan data:
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");
Beberapa kolom khusus, nilai pertamanya tidak boleh dihitung oleh suatu fungsi. Kita dapat menentukan bahwa ini adalah yang pertama di baris kolom [`numTrades] - jika memiliki 0, maka nilainya adalah yang pertama. Q memiliki fungsi pemilihan -? [Daftar Boolean; list1; list2] - yang memilih nilai dari daftar 1 atau 2 tergantung pada kondisi dalam argumen pertama:
Di sini saya memanggil tugas umum dengan fungsi saya (ekspresi dalam kurung kurawal). Nilai saat ini (argumen pertama) dan argumen tambahan, yang saya sampaikan di parameter ke-4, diteruskan ke sana.
Secara terpisah, kami menambahkan speaker baterai, karena bagi mereka fungsinya sama:
Ini adalah penugasan biasa dengan standar Q, saya hanya menetapkan daftar nilai sekaligus. Akhirnya, buat fungsi utama:
Dengan ungkapan ini, saya secara dinamis membuat fungsi dari string yang berisi ekspresi yang saya kutip di atas. Hasilnya akan terlihat seperti ini:
{[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])]}
Urutan perhitungan kolom terbalik, karena dalam Q urutan perhitungan dari kanan ke kiri.
Sekarang kami memiliki dua fungsi utama yang diperlukan untuk perhitungan, tetap menambahkan sedikit infrastruktur dan layanan siap.
Langkah terakhir
Kami memiliki fungsi preprocess dan updateAgg yang melakukan semua pekerjaan. Tetapi masih perlu untuk memastikan transisi yang benar dalam hitungan menit dan menghitung indeks untuk agregasi. Pertama kita mendefinisikan fungsi init:
init:{ tradeAgg:: 0#enlist[initWith];
Kami juga menetapkan fungsi roll, yang akan mengubah menit saat ini:
roll:{[tm] if[currTime>tm; :init[]];
Kami membutuhkan fungsi untuk menambahkan karakter baru:
addSyms:{[syms] currSyms,::syms;
Dan akhirnya, fungsi pembaruan (nama tradisional fungsi ini untuk layanan Q), yang dipanggil oleh klien, untuk menambahkan data:
upd:{[tblName;data]
Itu saja. Berikut adalah kode lengkap layanan kami, seperti yang dijanjikan, hanya beberapa baris:
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]; };
Pengujian
Periksa kinerja layanan. Untuk melakukan ini, jalankan dalam proses terpisah (masukkan kode dalam file service.q) dan panggil fungsi init:
q service.q –p 5566 q)init[]
Di konsol lain, mulai proses Q kedua dan sambungkan ke yang pertama:
h:hopen `:host:5566 h:hopen 5566
Pertama, buat daftar karakter - 10.000 lembar dan tambahkan fungsi untuk membuat tabel acak. Di konsol kedua:
syms:`IBM`AAPL`GOOG,-9997?`8 rnd:{[n;t] ([] sym:n?syms; time:t+asc n#til 25; price:n?10f; size:n?10)}
Saya menambahkan tiga karakter nyata ke daftar karakter untuk membuatnya lebih mudah untuk mencarinya di tabel. Fungsi rnd membuat tabel acak dengan n rows, di mana waktu bervariasi dari t hingga t + 25 milidetik.
Sekarang Anda dapat mencoba mengirim data ke layanan (tambahkan sepuluh jam pertama):
{h (`upd;`trade;rnd[10000;x])} each `time$00:00 + til 60*10
Anda dapat memeriksa dalam layanan bahwa tabel telah diperbarui:
\c 25 200 select from tradeAgg where sym=`AAPL -20#select from tradeAgg where sym=`AAPL
Hasil:
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
Kami sekarang akan melakukan pengujian beban untuk mengetahui berapa banyak data yang dapat diproses layanan per menit. Biarkan saya mengingatkan Anda bahwa kami menetapkan interval untuk pembaruan menjadi 25 milidetik. Oleh karena itu, suatu layanan harus (rata-rata) sesuai dengan setidaknya 20 milidetik per pembaruan untuk memberi pengguna waktu untuk meminta data. Masukkan yang berikut ini di proses kedua:
tm:10:00:00.000 stressTest:{[n] 1 string[tm]," "; times,::h ({st:.zT; upd[`trade;x]; .zT-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 adalah dua menit. Anda dapat mencoba memulai lebih dari 1000 baris setiap 25 milidetik:
start 1000
Dalam kasus saya, hasilnya adalah sekitar beberapa milidetik per pembaruan. Jadi saya akan segera menambah jumlah baris menjadi 10.000:
start 10000
Hasil:
min| 00:00:00.004 avg| 9.191458 med| 9f max| 00:00:00.030
Sekali lagi, tidak ada yang istimewa, tetapi ini adalah 24 juta baris per menit, 400 ribu per detik. Selama lebih dari 25 milidetik, pembaruan melambat hanya 5 kali, tampaknya saat mengubah menit. Tingkatkan menjadi 100.000:
start 100000
Hasil:
min| 00:00:00.013 avg| 25.11083 med| 24f max| 00:00:00.108 q)sum times 00:02:00.532
Seperti yang Anda lihat, layanan ini hampir tidak bisa, tetapi tetap berhasil bertahan. Jumlah data ini (240 juta baris per menit) sangat besar, dalam kasus seperti itu biasa menjalankan beberapa klon (atau bahkan puluhan klon) dari layanan, yang masing-masing hanya memproses sebagian karakter. Namun demikian, hasilnya mengesankan untuk bahasa yang ditafsirkan, yang terutama difokuskan pada penyimpanan data.
Mungkin timbul pertanyaan, mengapa waktu tumbuh non-linear dengan ukuran setiap pembaruan. Alasannya adalah bahwa fungsi pemerasan sebenarnya adalah fungsi C yang bekerja jauh lebih efisien daripada updateAgg. Dimulai dengan beberapa ukuran pembaruan (sekitar 10.000), pembaruanAgg mencapai puncaknya dan kemudian waktu pelaksanaannya tidak tergantung pada ukuran pembaruan. Ini karena langkah awal Q bahwa layanan ini dapat mencerna volume data seperti itu. Ini menekankan betapa pentingnya ketika bekerja dengan data besar untuk memilih algoritma yang tepat. Poin lainnya adalah penyimpanan data yang benar dalam memori. Jika data tidak disimpan dalam kolom atau tidak dipesan berdasarkan waktu, maka kita akan berkenalan dengan hal seperti miss cache TLB - tidak adanya alamat halaman memori dalam cache alamat prosesor. Menemukan alamat membutuhkan waktu sekitar 30 kali lebih lama jika terjadi kegagalan dan dalam kasus data yang tersebar dapat memperlambat layanan beberapa kali.
Kesimpulan
Dalam artikel ini, saya menunjukkan bahwa basis data KDB + dan Q cocok tidak hanya untuk menyimpan data besar dan akses mudah ke mereka melalui pilih, tetapi juga untuk membuat layanan pemrosesan data yang dapat mencerna ratusan juta baris / gigabyte data bahkan dalam satu proses Q tunggal . Bahasa Q itu sendiri memungkinkan penerapan algoritma yang berhubungan dengan pemrosesan data yang sangat singkat dan efisien karena sifat vektornya, penafsir bawaan dari dialek SQL, dan serangkaian fungsi perpustakaan yang sangat berhasil.
Saya akan perhatikan bahwa di atas hanya sebagian dari kemampuan Q, ia memiliki fitur unik lainnya. Misalnya, protokol IPC yang sangat sederhana yang menghapus batas antara proses Q individual dan memungkinkan Anda untuk menggabungkan ratusan proses ini ke dalam jaringan tunggal, yang dapat ditemukan di lusinan server di berbagai belahan dunia.