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.

[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. 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.

from tkati_core import LoopStats

PHASES = ("read", "produce", "commit")
stats = LoopStats(name="my-node", phases=PHASES)

while True:
    with stats.phase("read"):
        batch = consumer.read_arrow(...)
    stats.iterations += 1
    if batch is None:
        stats.starved_iterations += 1
        continue
    stats.rows_in += len(batch)

    with stats.phase("produce"):
        producer.produce_arrow(batch)
    stats.rows_out += len(batch)

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

    stats.report_if_due()

my-node perf over 10s: 157000 rows in, 153880 out (3120 dropped), 157 iterations (0 input-starved)
my-node perf: read=4.91s (49%) produce=3.96s (39%) commit=0.38s (4%)

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: a read phase usually 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.

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.

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.

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
table = consumer.read_arrow(
    aggregation_interval_seconds=settings.input.consumer.batch_timeout_sec,
    max_events_to_aggregate=settings.input.consumer.batch_size,
)
consumer.commit()

The factory method sets enable.auto.commit=False — offsets must be committed explicitly via consumer.commit().

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.

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:

[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.