Kdb pykx Wie aktiviere ich parallele Lesevorgänge (und erlauben trotzdem Schreibvorgänge)? ⇐ Python

[Anonymous]

Ich erstelle einen Fastapi -Service, der über Pykx mit KDB+/Q spricht. Derzeit dient es Endpunkte, die Abfragen über eingebettete q in einer inmemory globalen variablen PYKX -Tabelle ausführen. Ich skaliere jetzt meine Anwendung, damit X -Leseabfragen gleichzeitig bearbeitet werden können (z. B. viele Anfragen, die gleichzeitig denselben Endpunkt durchführen, um verschiedene Aggregationen auszuführen, aber ohne Schreiben.). Ich muss immer noch gelegentliche Schreibvorgänge unterstützen (z. B. Laden/Aktualisieren einer Tabelle im Speicher). Wenn ich jedoch mehrere asynchronisierte Abfragen aus Python abfelle, scheint der Q -Server sie seriell zu verarbeiten?

Code: Select all

# simple test server
q server.q -p 5000

< /code>
import asyncio
import pykx as kx

qconn_async = kx.AsyncQConnection('127.0.0.1', 5000)

async def run_query(arg: int):
return await qconn_async .qsql.select("my_table", where=[kx.Column("my_col")==arg])

async def main():
args = [5_000_000, 5_000_000, 5_000_000]
results = await asyncio.gather(*(run_query(a) for a in args))
print(results)

asyncio.run(main())

< /code>
[b]Observation[/b]: Even when I issue queries concurrently (via asyncio.gather), they appear to be processed serially on the server—total wall-time is roughly the sum of each query’s time.
I have also tried running mserve.q from the docs (https://code.kx.com/q/kb/load-balancing/

): https://github.com/kxsystems/kdb/blob/e/mserve.q

aber immer noch kein Glück. /> In der Lage, Q-Tabellen in einen Prozess zu schreiben (ich akzeptiere Inkonsistenz zwischen den Lesern, während Daten geschrieben werden) < /li>
< /ul>
Bearbeiten: Update (Sep 30, 2025): Versuchs Multi-Thread-Eingang + Timings < /strong> < /p>
Server -Setup < /p>
# startup.q creates a 10M-row table `my_table`
q startup.q -p -5000 / negative port = multi-threaded input??
< /code>
Client test (PyKX / asyncio)
import asyncio
import pykx as kx

qconn_async = kx.AsyncQConnection('127.0.0.1', 5000)

async def run_query(arg: int):
# simple full-table read for timing
res = await qconn_async.qsql.select("my_table")
return res

async def main():
tasks = [run_query(i) for i in range(5)]
results = await asyncio.gather(*tasks)

asyncio.run(main())
< /code>
Observation: total wall time scales ~linearly with the number of concurrent queries:
1 query: ~3.32s
2 queries: ~6.36s
5 queries: ~13.8s