Skip to content

Import Data into KDB-X DB Service

This page introduces the data import capabilities of KDB-X DB Service. It explains the available ingestion methods, how import jobs are processed, and how to choose the appropriate approach for batch, inline, or database-based ingestion workflows.

The KDB-X DB Service supports several data ingest methods.

  • File import, via API: batch ingestion from delimited text (CSV), Parquet, or q binary. Use this for running backfills or scheduled imports from file.

  • kdb database import, via API: ingest an existing kdb+ HDB. Use this for migrating partitioned historical data from an existing kdb+ database, or if you have complex transform needs on large datasets. The import is highly performant, it does the least amount of processing of data.

  • API data import, via API: that inserts rows directly via API request. Use this when inserting small volumes of data, integrating with applications, or bootstrapping a dataset. Do not use this for large batches of data, or high-frequency streaming data imports — kdb database import or streaming ingest methods are more performant.

  • Streaming ingest, via Reliable Transport (RT): high performance streaming ingest.

Importing an existing kdb+ database

The methods on this page ingest data into a running DB Service. To migrate an existing partitioned kdb+ HDB into a new clustered deployment, use initial import instead — it registers the staged data in place using symbolic links rather than re-ingesting it, which is substantially faster for large datasets.

Detailed documentation, including examples, for all APIs is available in the OpenAPI documentationc. Sample notebooks demonstrating API usage are bundled with the DB Service repo on GitHub.

Timezones

All timestamps are considered UTC in the KDB-X DB Service. Timestamps can however be queried and converted for other timezones.

File staging

Before importing files, they must be accessible to the DB Service in the appropriate staging location.

Place files in the data/imports/ directory relative to the Docker Compose working directory. The init-db.sh script pre-populates this directory with sample data.

cp mydata.csv data/imports/

Staging works the same way as single-node, but on a persistent volume rather than local disk. Each shard has its own /imports directory, backed by a persistent volume claim, on the Storage Manager pod of that shard. Copy the file into the directory of the shard you are importing into:

kubectl cp ./mydata.csv \
  "${NAMESPACE}/${RELEASE_NAME}-db-sm-0:/imports/mydata.csv" \
  -c sm

As in single-node, the path in the import request is relative to the staging directory — so the copy above is imported as "path": "mydata.csv".

Assembly selection

Each assembly has its own independent set of tables, so an import job always targets a single assembly. On a deployment configured with more than one assembly, identify the target with the assembly query parameter.

Job control

All import requests submitted through the API are asynchronous and return a jobId, which can be used to check status and cancel the job. Streaming ingest is continuous rather than job-based, and is not covered by these endpoints.

File import

The file import endpoint is POST /api/v0/imports/files. The API supports delimited text (CSV), Parquet, and q binary (as a single table in a file). Files are processed as follows:

  1. Data is read and parsed from specified files in the staging location. Import paths support globbing:
    • * matches within a path; for example, January/USD*.csv
    • ** matches across subdirectories; for example, forex/**/quote*.csv
  2. Transformations specified in the postparse dictionary are applied (delimited text only). Postparse applies basic q expressions to columns, and can:
    • Modify values of existing columns; for example, to handle a tricky timestamp format
    • Create entirely new derived columns
    • Reference values from one or more other columns
    • Include the filename as a value in the expression
  3. Columns specified for include are carried forward (delimited text only).

Data is read in chunks, and processed data is written to an intermediate location. Once all specified files are processed, the intermediate data is efficiently merged into the database.

If a target table does not exist, it can be created by setting the createTable flag on API imports. For production use, it is recommended to create tables explicitly with the table management API to maintain full control over table configuration.

Clustered deployments

createTable does not work in a clustered deployment. Tables must be defined in the assembly YAML before importing into them.

For help diagnosing column or type mismatches between incoming data and an existing table definition, see Troubleshooting.

For date partitions that already exist, the mode setting controls how new data is handled - either merge or overwrite for that date's partition.

Delimited text (CSV) import options

Several optional parameters are available to control parsing of delimited text:

  • delimiter: character that separates fields in the file. Default is a comma ","
  • decimal: character that is the decimal separator. Default is a period "."
  • header: list of strings that specify (or override) column names for the file.
  • headerRowIndex: index (from 0) to start reading the header. -1 indicates no header; values greater than 0 will skip initial rows (for example, comments)
  • types: data types string, in Tok format (uppercase).

Example file import

An example CSV import that uses multiple features at once:

session.importFiles([
  table:"fxquote";
  path:"fxquote.csv.gz";
  delimiter:",";
  decimal:".";
  header:("trddate";"ts";"sym";"bid";"ask");
  types:"DPSFF";
  postparse:([sprd:"(data`ask)-data`bid";
              bid:"data`bid";
              ask:"data`ask"]);
  include:("trddate";"ts";"sym";"bid";"ask";"sprd");
  createTable:1b]);

session.import_files(
  table="fxquote",
  path="fxquote.csv.gz",
  delimiter=",",
  decimal=".",
  header=["trddate","ts","sym","bid","ask"],
  types="DPSFF",
  postparse={"sprd":"(data`ask)-data`bid",
          "bid":"data`bid",
          "ask":"data`ask"},
  include=["trddate","ts","sym","bid","ask","sprd"],
  createTable=True);

curl -s -X POST "http://localhost:8080/api/v0/imports/files" \
  -H "Accept: application/json" \
  -H "Content-Type: application/json" \
  -d '{"table":"fxquote",
      "path":"fxquote.csv.gz",
      "delimiter":",",
      "decimal":".",
      "header":["trddate","ts","sym","bid","ask"],
      "types":"DPSFF",
      "postparse":{"sprd":"(data`ask)-data`bid",
                  "bid":"data`bid",
                  "ask":"data`ask"},
      "include":["trddate","ts","sym","bid","ask","sprd"],
      "createTable":true}'

kdb database import

The kdb database import endpoint is POST /api/v0/imports/kdb. The API allows you to import an existing partitioned kdb+ HDB, or splayed kdb+ tables. This is particularly useful for migrating historical databases from another kdb+ system.

As the most efficient method for merging data into the KDB-X DB Service, it is also useful for large batch imports or batch data that requires more complex transformations than those supported by the file import API. In these cases, data can be persisted to a kdb+ HDB as the final step of an extract/transform pipeline, and this HDB can then be loaded using the kdb database API.

Partitioned HDBs must be partitioned by date on a timestamp column.

Example kdb database import

session.importKDB([table:"fxquote";path:"fxquote-hdb";mode:"overwrite"])

session.import_database(table="fxquote", path="fxquote-hdb", mode="overwrite")

curl -s -X POST "http://localhost:8080/api/v0/imports/kdb" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json" \
  -d '{
      "table": "fxquote",
      "path": "fxquote-hdb",
      "mode": "merge"}'

API data import

JSON import vs. Streaming

Importing data via JSON API is not efficient for large batch loads (use file or kdb database import instead), or for streaming real-time data (use streaming ingest instead). It is best used for small and infrequent regular updates, or as a convenient ad-hoc manual import.

The JSON import endpoint is POST /api/v0/imports/data. The API allows you to import JSON data directly into the database. Data can be sent in two different formats:

  • Rows as lists, matching the table schema: 
    ["2026-01-21", "2026-01-21T10:00:00.000", "EURUSD", 901.2, 901.3],
["2026-01-21", "2026-01-21T10:00:00.000", "EURUSD", 901.2, 901.3]
  • Objects with explicitly named fields: 
    {"instrumentid": 77, "sym": "USDBRL", "category": "EM", "decimals": 4, "pipdecimals": 4},
{"instrumentid": 78, "sym": "USDKRW", "category": "EM", "decimals": 2, "pipdecimals": 2}

As with CSV import, additional options are available to help resolve column names and data types:

  • columnNames: specifies column names for row payloads.
  • types: data types string, in cast format (lowercase).

Example JSON data import

The following example uses the row format:

// Rows payload imported to the 'fxquote' table. ColumnNames are required for row inserts.
data:(("2026-01-21";"2026-01-21T10:00:00.000";"EURUSD";901.2;901.3);("2026-01-21";"2026-01-21T10:00:00.000";"EURUSD";901.2;901.3));
session.importData[`table`data`columnNames!(`fxquote;data;(`trddate`ts`sym`bid`ask))]

// Objects payload imported to 'instruments' table
data:((`instrumentid`sym`category`decimals`pipdecimals)!(77;"USDBRL";"EM";4;4);(`instrumentid`sym`category`decimals`pipdecimals)!(78;"USDKRW";"EM";2;2));
session.importData[`table`data!(`instruments;data)]

# Rows payload imported to the 'fxquote' table. ColumnNames are required for row inserts.
session.import_data(
    table="fxquote",
    data=[
        ["2026-01-21", "2026-01-21T10:00:00.000", "EURUSD", 901.2, 901.3],
        ["2026-01-21", "2026-01-21T10:00:00.000", "EURUSD", 901.2, 901.3],
    ],
    columnNames=["trddate", "ts", "sym", "bid", "ask"],
    insert_as="rows",
)

# Objects payload imported to 'instruments' table
session.import_data(
    table="instruments",
    data=[
        {"instrumentid": 77, "sym": "USDBRL", "category": "EM", "decimals": 4, "pipdecimals": 4},
        {"instrumentid": 78, "sym": "USDKRW", "category": "EM", "decimals": 2, "pipdecimals": 2},
    ],
    insert_as="objects",
)

# Rows payload imported to the 'fxquote' table. ColumnNames are required for row inserts.
curl -s -X POST "http://localhost:8080/api/v0/imports/data" \
-H "Accept: application/json" \
-H "Content-Type: application/json" \
-d '{
    "table": "fxquote",
    "data": [
    ["2026-01-21", "2026-01-21T10:00:00.000", "EURUSD", 901.2, 901.3],
    ["2026-01-21", "2026-01-21T10:00:00.000", "EURUSD", 901.2, 901.3]
    ],
    "columnNames": ["trddate", "ts", "sym", "bid", "ask"]
}'

# Objects payload imported to 'instruments' table
curl -s -X POST "http://localhost:8080/api/v0/imports/data" \
-H "Accept: application/json" \
-H "Content-Type: application/json" \
-d '{
    "table": "instruments",
    "data": [
    {"instrumentid": 77, "sym": "USDBRL", "category": "EM", "decimals": 4, "pipdecimals": 4},
    {"instrumentid": 78, "sym": "USDKRW", "category": "EM", "decimals": 2, "pipdecimals": 2}
    ]
}'

Streaming ingest

Data can be streamed into the database using Reliable Transport (RT), a high performance data ingest system. Updates should be streamed to the data topic.

The single-node DB Service incorporates a single-node deployment of RT. By default, the RT endpoint is available on port 5002.

{
  "name": "dbs-fxpub",
  "useSslRt": false,
  "topics": { "insert": "data" },
  "insert": { "insert": [":localhost:5002"] }
}

Each shard has its own RT stream, served by its own RT service — named rt-<release-name>-0 — on port 5002. Update the insert address in your RT config to point at the RT service of the shard you are publishing to.

If the RT service is not exposed outside the cluster, port-forward it first:

kubectl port-forward "svc/rt-${RELEASE_NAME}-0" 5002:5002 -n "$NAMESPACE" &

{
  "name": "dbs-fxpub",
  "useSslRt": false,
  "topics": { "insert": "data" },
  "insert": { "insert": [":rt-my-shard-release-0:5002"] }
}

Example data feed

Prerequisite

Before running the sample feed, ensure that the fxquote table already exists in the database. The sample feed inserts data into an existing table and does not create it automatically. If you create the table manually, define the sym column as a symbol type.

An example Python feed is included with the DB Service, reproduced below.

from rtpy import rt_helper
from datetime import datetime, timezone
from random import random, choice
import time
import os

print("Starting feed...", flush=True)

symlist = {'EURUSD':[1.16,4],'GBPUSD':[1.34,4],'USDJPY':[158,2]}

cfg_path = 'file://' + os.path.dirname(os.path.realpath(__file__)) + '/rtconfig.json'
params = rt_helper.RTParams(config_url=cfg_path, console_log_level='error')
h, status_code = rt_helper.start(params)
print("Feed started; publishing to fxquote. Press Ctrl+C to stop.", flush=True)

published = 0
while True:
    for sym in symlist:
        mid,dec = symlist[sym]
        newmid = round(mid + choice([-1,1]) * random()/10**(dec), dec)
        spread = round(random()/10**(dec-1),dec)
        now = datetime.now(timezone.utc)
        price = [{'trddate':now.date(),
                  'ts':now,
                  'sym': sym,
                  'bid': round(newmid-spread/2,dec+1),
                  'ask': round(newmid+spread/2,dec+1)}]
        symlist[sym][0] = newmid
        rt_helper.insert(h, 'fxquote', price)
        published += len(price)
    if published % 30 == 0:
        print(f"Published {published} rows to fxquote.", flush=True)    
    time.sleep(1)

The feed reads its rtconfig.json from the same directory as the script. The configuration for each deployment type is shown under Streaming ingest above.

RT interfaces are available for C/C++, Java, q, C#, and Python. Most production feed solutions use the C/C++ or Java SDKs.

Run the sample feed in a separate terminal from the running DB Service or client session. Alternatively, run the feed in the background with python3 fxfeed.py &.

The feed runs continuously, publishing FX quote data to the fxquote table until you stop it with Ctrl+C. It prints a confirmation message when it starts.

To verify that data is being published, run the following row-count query while the feed is active. Run the query again after a few seconds and confirm that the rowCount value has increased.

curl -s -X POST "http://localhost:8080/api/v0/query/q" \
  -H "Accept: application/json" \
  -H "Content-Type: application/json" \
  -d '{"query": "select rowCount:count i from fxquote"}'

Next steps