Publish data from a Kafka topic
Ingest a topic continuously
A Kafka source uses @stream_publisher rather than @publisher. A long-lived worker consumes
the topic, and each time a rollover threshold trips it writes the completed buffer out and the
publisher commits it as a new table version. One source reads one topic, and the topic name is
also the table name.
There are several ways to pass secrets into Tabsdata. See Secrets Management for more info if you are planning to have an LLM register secrets into Tabsdata's credential store.
Listed 1 directory, called tabsdata 3 times (ctrl+o to expand)
The Kafka publisher guide and connection template are loaded. This one is a stream publisher, so it also needs to know when to roll a batch over. Five values left, then I generate and register. Reply with them one per line, or point me at a file that has them:
- collection name
- bootstrap servers and SASL credentials
- topic
- message format and schema
- rollover thresholds
Read 2 files, called tabsdata 2 times (ctrl+o to expand)
SASL password found, and the Avro schema read from ~/page_views.avsc. Reading back what is not secret: broker.internal:9092, topic page_views, format avro, dead letter topic page_views_dlq, collection events_src.
Called tabsdata 4 times (ctrl+o to expand)
Collection events_src created in group sources, with a KafkaStreamSrcConn to broker.internal:9092. Connection check passed, and the topic exists with 6 partitions.
Called tabsdata 2 times (ctrl+o to expand)
The stream publisher, registered in events_src. flush_interval_mins is a ceiling rather than a schedule: whichever of it and the connector thresholds trips first ends the batch, so the 128 MB and 300 second limits usually fire before the 5 minute fallback:
@stream_publisher(
source=KafkaStreamSrc(
topic="page_views",
data_format="avro",
schema=PAGE_VIEWS_SCHEMA,
dlq_topic="page_views_dlq",
src_cfg={
"kafka.time_rollover_seconds": 300,
"kafka.size_rollover_mb": 128,
},
),
flush_interval_mins=5,
output_tables=["page_views"],
)
def read_page_views(page_views: TableFramesSpec) -> TableFrameSpec:
return concat(page_views)
Called tabsdata 6 times (ctrl+o to expand)
Stream worker started. First rollover committed after 128 MB, 0 failures.
events_src/page_views is live and keeps gaining versions, one per rollover:
┌───────────────────┬───────────────┐ │ field │ value │ ├───────────────────┼───────────────┤ │ collection │ events_src │ ├───────────────────┼───────────────┤ │ table │ page_views │ ├───────────────────┼───────────────┤ │ rows this version │ 1,904,338 │ ├───────────────────┼───────────────┤ │ columns │ 14 │ ├───────────────────┼───────────────┤ │ rollover trigger │ size (128 MB) │ ├───────────────────┼───────────────┤ │ offsets committed │ yes │ └───────────────────┴───────────────┘
schema is required, and messages that fail to deserialize go to dlq_topic instead of
failing the batch. A stream publisher takes no trigger_by at all, and it must not declare a
ctx parameter, because it runs inside the stream loop rather than once per transaction.