docs.fermihdi.io — python.md

/ › hd › 2.0.4 › md › sdk

Download python.mdView raw

FermiHDI SDK — Python

BETA. Builds and links against a real SDK on every change, and the module imports. It has not been exercised against a live cluster as heavily as the Go binding.

pybind11 bindings over the C++ SDK. Read the SDK overview first — especially §1, the control plane / data plane split.

Install

The module is not on PyPI: it is a compiled extension that links the SDK, and it ships in the SDK package image alongside the library it needs. Copy the package into your image and put its Python directory on PYTHONPATH:

COPY --from=cr.fermihdi.io/hd/sdk:<version> /opt/fermihdi-sdk /opt/fermihdi-sdk
ENV PYTHONPATH=/opt/fermihdi-sdk/python LD_LIBRARY_PATH=/opt/fermihdi-sdk/lib

The directory holds fermihdi_sdk.cpython-*.so, built for the package's CPU variant and Python version (PACKAGE.txt records both). Match the variant to the host the client runs on: an avx512 module raises "Illegal instruction" on a machine without AVX-512.

You will need the TLS tunable

Python dlopens the extension, and the SDK carries the HD engine runtime's initial-exec thread-local storage. Without surplus static TLS the import fails:

ImportError: ... cannot allocate memory in static TLS block

GLIBC_TUNABLES=glibc.rtld.optional_static_tls=1048576 python3 your_app.py

This is a property of the SDK, not of this binding — see §5.3.

API

FermiHDIQueryClient

FermiHDIQueryClient(proxy_url="", data_plane_iface="", ctrl_plane_iface="")

  stop()
  register_with_dms(dms_url, node_name, override_key_pem="") -> RegistrationResult
  add_control_header(header)
  set_scc_targets(targets)                 # [SccAddress]
  load_dataset_schema(schema_json)
  generate_batch_ids_time(from_sec, to_sec) -> [str]
  generate_hd_query(hd_query) -> str
  get_found_batch_ids() -> [str]
  get_missing_batch_ids() -> [str]
  prepare_query(query_id)                  # arm the listener BEFORE you dispatch
  set_expected_batches(batch_ids)          # the proxy's found_batches, AFTER
  set_filter_verdict(would_exclude_batches, bloomrf_mode)
  set_filter_verdict_json(verdict_json)    # the proxy's result event, verbatim
  get_filter_false_negatives() -> [str]    # must be empty
  get_bloomrf_mode() -> str
  set_local_aggregations(agg_config)
  execute_dataset(sql_or_json) -> ArrowTable      # the raw record table, untouched
  execute_dataset_count(sql_or_json) -> int
  execute_to_sql(fermihdi_json, duckdb_sql) -> ArrowTable
  execute_sql_on_empty(duckdb_sql) -> ArrowTable   # the SQL over zero records
  apply_sql(table, duckdb_sql) -> ArrowTable       # SQL over a table you hold

ArrowTable                    # what every execute_* returns
  num_rows() num_columns() schema()
  to_ipc() -> bytes           # pyarrow.ipc.open_stream(b).read_all()
table_from_ipc(bytes) -> ArrowTable   # e.g. from pyarrow: sink.getvalue().to_pybytes()
  execute_vector_search(sql, vector_column, target, top_k) -> pyarrow.Table
  execute_graph(sql, src_col, dst_col)

Supports the context-manager protocol (__enter__ / __exit__), which is the recommended way to use it — stop() is then guaranteed.

FermiHDIIngestClient

FermiHDIIngestClient(data_plane_iface="", ctrl_plane_iface="")

  register_with_dms(...)          discover_scc_targets(proxy_url) -> bool
  load_dataset_schema(json)       set_scc_targets(targets)
  generate_batch_id_string(s)     generate_batch_id_number(n)
  generate_batch_id_time(sec)     convert_record_to_binary(record)
  append_record_to_arrow(record)  flush_arrow_batch(batch_id)
  stream_insert(batch_id, batch, broadcast=False) -> bool
  send_eob(batch_id)              send_bbn(batch_id)

Types

Filter(field_name, value, op)     # op: FilterOp.EQUALITY | NOT_EQUAL | GREATER | LESS
SubQuery(filters=[...])
SccAddress(ip, port)
HDQuery(query_id, batch_ids, sub_queries, targets,
        query_qos=1, acknowledge_full_dataset_scan=False)

Worked example

import os, json, secrets
import fermihdi_sdk as fsdk

schema = json.dumps({"datasets": [{"record_len": 89, "time_index_interval": 60}]})

with fsdk.FermiHDIQueryClient("https://proxy.internal:443") as client:
    client.load_dataset_schema(schema)

    batch_ids = client.generate_batch_ids_time(1788500000, 1788503600)

    hq = fsdk.HDQuery()
    hq.query_id = secrets.randbelow(2**32)     # distinct per query
    hq.batch_ids = batch_ids
    hq.targets = [fsdk.SccAddress("10.20.0.14", 6987)]

    sq = fsdk.SubQuery()
    # An address is its integer: 10.0.0.1 is 167772161. Convert before the SDK sees it.
    sq.filters = [fsdk.Filter("ip_src", "167772161", fsdk.FilterOp.EQUALITY)]
    hq.sub_queries = [sq]

    body = client.generate_hd_query(hq)
    if not body:
        raise RuntimeError("refused: open query without acknowledge_full_dataset_scan")

    table = client.execute_to_sql(body, "SELECT ip_src, bytes FROM ipflows LIMIT 100")
    print(table.num_rows, "rows")

Use cases

Notebook analysis. The context manager plus execute_to_sql returning a pyarrow.Table drops straight into pandas:

df = table.to_pandas()

Vector search over stored embeddings.

hits = client.execute_vector_search(body, "embedding", target_vector, top_k=20)

A counting query, when you want cardinality without materialising rows:

n = client.execute_dataset_count(body)

Ingest from a Python producer. Build records, append to an Arrow batch, flush under a batch ID, then mark the batch complete:

ing = fsdk.FermiHDIIngestClient()
ing.load_dataset_schema(schema)
ing.discover_scc_targets("https://proxy.internal:443")
bid = ing.generate_batch_id_time(1788500000)
for rec in records:
    ing.append_record_to_arrow(rec)
ing.flush_arrow_batch(bid)
ing.send_eob(bid)

Gotchas

Download python.mdView raw