Publish data from a Kafka topic
Ingest a topic continuously
A Kafka source lets a stream publisher ingest a Kafka topic into a Tabsdata table as messages arrive, handing them to the function in batches.
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 hand over a batch. Six 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
- consumer group
- topic
- message format and schema
- when to hand over a batch
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, group tabsdata_page_views, 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. The topic exists with 6 partitions.
Called tabsdata 2 times (ctrl+o to expand)
The stream publisher, registered in events_src. A batch is due after 5 minutes, or sooner once 128 MB are waiting, but never sooner than a minute after the previous one:
@stream_publisher(
source=KafkaStreamSrc(
topic="page_views",
data_format="avro",
schema=PAGE_VIEWS_SCHEMA,
dlq_topic="page_views_dlq",
),
batch_max_secs=300,
batch_max_bytes=128 * 1024 * 1024,
output_tables=["page_views"],
)
def read_page_views(page_views: TableFramesSpec) -> TableFrameSpec:
return concat(page_views, how="diagonal_relaxed")
Called tabsdata 6 times (ctrl+o to expand)
Stream worker started. First batch committed after 128 MB, 0 failures.
events_src/page_views is live and keeps gaining versions, one per batch:
┌───────────────────┬───────────────┐ │ field │ value │ ├───────────────────┼───────────────┤ │ collection │ events_src │ ├───────────────────┼───────────────┤ │ table │ page_views │ ├───────────────────┼───────────────┤ │ rows this version │ 1,904,338 │ ├───────────────────┼───────────────┤ │ columns │ 14 │ ├───────────────────┼───────────────┤ │ batch due on │ size (128 MB) │ ├───────────────────┼───────────────┤ │ offsets committed │ yes │ └───────────────────┴───────────────┘
schema is required, and messages that fail to deserialize go to dlq_topic while the stream
carries on. Without a dlq_topic, such a message stops the stream. 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.
See Kafka for every option of the source and the connection, and Batch parameters for when a batch is due.