Metadata-Version: 2.4
Name: tkati-core
Version: 0.6.0
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Distributed Computing
Requires-Dist: clickhouse-connect>=0.11.0
Requires-Dist: confluent-kafka>=2.11.0
Requires-Dist: loguru>=0.7.3
Requires-Dist: orjson>=3.9.0
Requires-Dist: prometheus-client>=0.21
Requires-Dist: pyarrow>=21.0.0
Requires-Dist: pydantic-settings>=2.11.0
Requires-Dist: tenacity>=9.0.0
Summary: Core abstractions for Kafka/ClickHouse streaming pipeline nodes
Keywords: kafka,clickhouse,streaming,pyarrow,data-pipeline
License: Apache-2.0
Requires-Python: >=3.13
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM
Project-URL: Repository, https://github.com/epoch8/tkati

# tkati-core

`tkati-core` provides the building blocks for streaming data pipeline nodes that
read from Kafka and write to Kafka or ClickHouse.

## Settings

Every backend's settings split into a **`connection`** tier (server-specific: how to
reach the broker/database) and a resource tier named for what that backend calls the
thing you read/write (`topic` for Kafka, `table` for ClickHouse) — plus, where relevant,
a tier for behavior local to this particular reader/writer (Kafka's `consumer` settings).
This keeps server-specific config separate from per-instance client config, and is
meant to stay consistent as more backends (e.g. RabbitMQ) are added.

```toml
[input]
type = "kafka"

[input.connection]
broker = "localhost:9092"

[input.topic]
# definition of input stream:
# - topic name
# - message schema
# - message format = "json" / "arrow-batch"

[input.consumer]
# parameters local to this consumer
# - group_id
# - batch_size
# - batch_timeout_sec
# - auto_offset_reset

[output]
type = "kafka"  # or "clickhouse"

[output.connection]
broker = "localhost:9092"

[output.topic]
# definition of output stream
# - topic name
# - message schema
# - message format = "json" / "arrow-batch"
# - key_column (optional) = column to use as the Kafka message key

[...]
# settings specific to node function
```

## Usage

### `Consumer` / `Producer` base classes

`tkati_core.consumer.Consumer` and `tkati_core.producer.Producer` are the abstract
interfaces a node's input and output are built against. A `Consumer` reads a batch
with `read_arrow` (an Arrow table, which fails the batch on a malformed message) or
`read_pylist` (a list of dicts, which skips and logs one). Either returns a
`ConsumedBatch`, whose `.data` is the table or list, or `None` when nothing arrived.
Each batch then ends in one of two explicit calls, made in the order the batches
were read:

- `consumer.commit(batch)`: the batch is fully processed. Exactly its offsets are
  committed, even if later batches have already been read.
- `consumer.rewind(batch)`: processing failed. The consumer seeks back to where the
  batch started, so it is read again, and so is everything read after it. Those
  later batches can no longer be committed.

Committing or rewinding anything but the oldest outstanding batch raises
`ValueError`. Nothing is ever committed implicitly. `KafkaConsumer` is the only
`Consumer` implementation today; `KafkaProducer` and `ClickhouseProducer` both
implement `Producer`. This is what lets a generic node pick its input/output kind
from config instead of hardcoding a concrete class.

### `LoopStats` — where a node's wall clock went

`tkati_core.stats.LoopStats` accumulates per-phase timings across a node's loop
and logs a breakdown on an interval (every 10s by default). The phase names,
the log prefix and the cadence are all constructor arguments, because nodes
have different pipelines — a dedup node has lookup and write phases an
extract/load node does not.

```python
from tkati_core import CONSUMER_PHASES, PRODUCER_PHASES, LoopStats

PHASES = (*CONSUMER_PHASES, *PRODUCER_PHASES, "commit")
stats = LoopStats(phases=PHASES)

while True:
    # Pass `stats` down and the consumer and producer time their own phases —
    # see below.
    batch = consumer.read_arrow(..., stats=stats)
    stats.iterations += 1
    if batch is None:
        stats.starved_iterations += 1
        continue
    stats.rows_in += len(batch.data)

    try:
        producer.produce_arrow(batch.data, stats=stats)
        producer.flush(stats=stats)
    except Exception:
        consumer.rewind(batch)
        raise
    stats.rows_out += len(batch.data)

    with stats.phase("commit"):
        consumer.commit(batch)

    stats.report_if_due()
```

```
perf over 10s: 157000 rows in, 153880 out (3120 dropped), 157 iterations (0 input-starved)
perf: consumer/poll=4.43s (44%) consumer/parse=0.48s (5%) producer/serialize=2.10s (21%) producer/enqueue=0.35s (4%) producer/deliver=1.51s (15%) commit=0.38s (4%)
```

`Consumer.read_arrow` and `Consumer.read_pylist` take an optional `stats`
and split their own time into the two phases named by `CONSUMER_PHASES`:
**`consumer/poll`**, fetching from the broker, and **`consumer/parse`**, turning
the raw payloads into an Arrow table or dicts. These have unrelated fixes
(batch sizing and broker latency versus JSON decoding cost), so they are worth
telling apart.

`Producer.produce_arrow`, `produce_pylist` and `flush` do the same with the
three `PRODUCER_PHASES`:

* **`producer/serialize`**: encoding rows into the wire format (the wire-type
  cast, `to_pylist` and `orjson.dumps`, or the Arrow IPC write for
  `arrow-batch`). This is CPU time in Python.
* **`producer/enqueue`**: the per-message `produce()` calls that hand the
  encoded bytes to librdkafka. It scales with message count, so `arrow-batch`
  (one message per batch) all but removes it.
* **`producer/deliver`**: waiting for the sink to accept. For Kafka that is
  `flush()`. librdkafka starts sending in the background during `enqueue`, so
  this is the *remaining* wait for broker acks, not the batch's total network
  time. `ClickhouseProducer` records its whole insert here, retries and DLQ
  fallback included, because `clickhouse_connect` encodes and sends in a
  single call.

The prefixes mark these figures as timed inside `tkati-core`. A node's own
phases stay unprefixed. Splice the tuples into your phase tuple rather than
writing the names out by hand, so a rename in core can't leave your column
silently reading `0.00s`.

Do **not** also wrap these calls in a phase of your own: that would count the
same time twice and break the invariant below.

`phases` is an explicit ordered tuple, not derived from which phases happened
to fire: a phase that doesn't run during an interval's first iteration would
otherwise shift the column order between reports, and a stable order is what
makes two consecutive lines comparable.

Percentages are of the interval rather than of each other, so they do **not**
sum to 100 — the shortfall is time in none of the named phases, which keeps
unaccounted work visible. Track `starved_iterations` for iterations that were
blocked waiting on input: `consumer/poll` blocks until the batch fills or the timeout
expires, so on an under-fed node it approaches 100% and nothing else on the
line means anything.

### Prometheus metrics

`tkati_core.metrics` exposes a `LoopStats` as Prometheus counters: the same
numbers as the log line, but monotonic, so they don't reset at each report.

```python
from tkati_core import MetricsSettings, start_metrics_server

stats = LoopStats(phases=PHASES)
start_metrics_server(MetricsSettings(), stats)  # :8000/metrics, daemon thread
```

| Metric | Labels | Meaning |
|---|---|---|
| `tkati_phase_seconds_total` | `phase` | wall clock spent in each phase |
| `tkati_wall_seconds_total` | — | wall clock since the `LoopStats` was created, the 100% denominator |
| `tkati_rows_in_total` / `tkati_rows_out_total` | — | rows read / written |
| `tkati_iterations_total` | — | loop iterations |
| `tkati_starved_iterations_total` | — | iterations that waited on input |

There is no node label: each node serves its own `/metrics`, so the scrape
target's `job`/`instance` already says which node a series came from. `phase`
is the phase name exactly as it appears in the log line. Every declared phase
is exported from the first scrape, at 0 if it hasn't run yet.

The log line's "% of the interval", and its unaccounted remainder, in PromQL:

```promql
rate(tkati_phase_seconds_total[1m])
  / ignoring(phase) group_left rate(tkati_wall_seconds_total[1m])

1 - sum without (phase) (rate(tkati_phase_seconds_total[1m]))
  / rate(tkati_wall_seconds_total[1m])
```

Wall clock is its own metric rather than a `phase="total"` series, because
`sum without (phase)` over phases would otherwise count it twice. Throughput is
`rate(tkati_rows_in_total[1m])`. The starved share is
`rate(tkati_starved_iterations_total[1m]) / rate(tkati_iterations_total[1m])`.

Values are read from `LoopStats.totals()` when Prometheus scrapes, so exporting
adds nothing to the loop. `MetricsSettings` (`enabled`, `port`, `addr`) is meant
to be embedded as a `metrics` section in a node's settings. It is on by default,
and `enabled = false` makes `start_metrics_server` a no-op. The collector,
`LoopStatsCollector`, can also be registered on your own `CollectorRegistry`
if you serve metrics yourself.

### `tkati_core.settings` — generic node settings aliases

`tkati_core.settings` defines `InputSettings`/`OutputSettings` (discriminated unions
over every input/output kind `tkati-core` implements). Use those aliases with the
factory helpers in `tkati_core.consumer` and `tkati_core.producer`, or import the
helpers from the top-level `tkati_core` package for convenience.

```python
from tkati_core import InputSettings, OutputSettings, build_consumer, build_producer
from tkati_core.settings import TomlBaseSettings

class AppSettings(TomlBaseSettings):
    input: InputSettings
    output: OutputSettings

settings = AppSettings()
consumer = build_consumer(settings.input)
producer = build_producer(settings.output)
```

`build_producer` also takes an optional `dlq_producer` kwarg, forwarded to
`ClickhouseProducer.from_output_settings` when `settings.type == "clickhouse"` (a no-op
for the `"kafka"` output kind, which has no DLQ-fallback logic of its own). The
recursive-split batch size for that fallback comes from `settings.dlq_split_factor`
(a field on `ClickHouseOutputSettings` itself), not from a separate parameter.

### Constructing a consumer from settings

Use `KafkaConsumer.from_input_settings` to construct a consumer directly from
`KafkaInputSettings` — no need to manually map fields to Confluent Kafka config keys.

```python
from tkati_core.settings import TomlBaseSettings
from tkati_core.kafka.settings import KafkaInputSettings
from tkati_core.kafka.consumer import KafkaConsumer

class AppSettings(TomlBaseSettings):
    input: KafkaInputSettings
    # ...

settings = AppSettings()  # settings.input.connection.broker, settings.input.topic.name, ...
consumer = KafkaConsumer.from_input_settings(settings.input)

# Read a batch
batch = consumer.read_arrow(
    timeout=settings.input.consumer.batch_timeout_sec,
    num_messages=settings.input.consumer.batch_size,
)
if batch is not None:
    process(batch.data)
    consumer.commit(batch)
```

The factory method sets `enable.auto.commit=False`: each batch's offsets are
committed explicitly with `consumer.commit(batch)`.

### Constructing a producer from settings

Use `KafkaProducer.from_output_settings` to construct a producer directly from
`KafkaOutputSettings`. It accepts PyArrow tables or record batches and handles
serialization according to the topic's `format` setting.

```python
from tkati_core.settings import TomlBaseSettings
from tkati_core.kafka.settings import KafkaOutputSettings
from tkati_core.kafka.producer import KafkaProducer

class AppSettings(TomlBaseSettings):
    output: KafkaOutputSettings
    # ...

settings = AppSettings()
producer = KafkaProducer.from_output_settings(settings.output)

# Produce a PyArrow table (one message per row for "json" format)
producer.produce_arrow(table)
producer.flush()
producer.close()  # flushes and releases resources
```

`ClickhouseProducer.from_output_settings` (in `tkati_core.clickhouse.producer`) works the
same way against `ClickHouseOutputSettings`.

**Formats** — controlled by `output.topic.format` in `settings.toml`:

- `"json"` *(default)*: each row becomes a separate Kafka message serialized with orjson.
- `"arrow-batch"`: the entire table is serialized as a single Arrow IPC stream message.

**Message keys** — controlled by `output.topic.key_column` in `settings.toml`:

```toml
[output.connection]
broker = "localhost:9092"

[output.topic]
name = "my-output-topic"
key_column = "customer_id"   # column whose value becomes the Kafka message key
```

`key_column` is optional. When omitted (or `None`), messages are produced without a key.
When set, the value of that column for each row is used as the Kafka message key
(JSON format only — ignored for `"arrow-batch"`). This determines which Kafka partition
each message is routed to.

