Publish data from a Kafka topic

Tabsdata ingests a Kafka topic through a streaming publisher: a @stream_publisher function whose source is KafkaStreamSrc, reading from a connection registered on the collection like any other connector.

How It Works

Staging

While the trigger is active, Tabsdata consumes messages from the topic and holds them in a private staging area.

Buffering

Messages accumulate in the stage until a rollover threshold is met: time, size, or message count.

Publishing

Once a threshold fires, the accumulated batch is passed to your function as a list of TableFrames, which you concatenate and return as a new table version.

Supported Formats

• Data formats: JSON, Avro, Protobuf.
• Schema management: provide the schema string directly, or resolve it from a Confluent Schema Registry or an AWS Glue Schema Registry.

Basic Configuration Example

The connection is registered on the collection, the same as any other connector:

tdk connection template --type tabsdatak.conn.kafka:KafkaStreamSrcConn --file conn-kafka.yaml
tdk collection create --name events --group sources --conn-file conn-kafka.yaml

conn-kafka.yaml

kind: connectionDef
apiVersion: '1.0'
type: tabsdatak.conn.kafka:KafkaStreamSrcConn
spec:
  servers: 'broker:9092'
  group_id: 'ingest'

• required
• required
• optional
• optional

The function only references KafkaStreamSrc; it never sees the connection, credentials, or schema registry directly:

import json

from tabsdatak.api import TableFrameSpec, TableFramesSpec, stream_publisher
from tabsdatak.conn.kafka import KafkaStreamSrc
from tabsdatak.tableframe.functions import concat

SCHEMA = json.dumps({
    "title": "FlightEvent",
    "type": "object",
    "properties": {"id": {"type": "integer"}, "status": {"type": "string"}},
    "required": ["id", "status"],
})


@stream_publisher(
    # The source only names the topic and payload format; the broker,
    # credentials, and schema registry all come from the collection's
    # connection.
    source=KafkaStreamSrc(
        topic="my_topic",
        data_format="json",
        schema=SCHEMA,
        src_cfg={"kafka.time_rollover_seconds": 30},
    ),
    flush_interval_mins=1,
    output_tables=["flight_events"],
)
def flight_events_publisher(events: TableFramesSpec) -> TableFrameSpec:
    # Concatenate the batch of messages into a single frame
    frames = [frame for frame in (events or []) if frame is not None]
    return concat(frames) if frames else None

Configuration Reference

KafkaStreamSrc

These parameters control how data is consumed and when it is committed to the table.

ingest.py

from tabsdatak.api import TableFrameSpec, TableFramesSpec, stream_publisher
from tabsdatak.conn.kafka import KafkaStreamSrc
from tabsdatak.tableframe.functions import concat

@stream_publisher(
  source=KafkaStreamSrc(
        topic="my_topic",
        data_format="json",
        schema=SCHEMA,
    flush_interval_mins=1,
    ),
  flush_interval_mins=1,
  output_tables=["my_topic"],
)
def ingest(events: TableFramesSpec) -> TableFrameSpec:
  frames = [f for f in (events or []) if f is not None]
  return concat(frames) if frames else None

• required
• required
• required
• optional
• optional
• required

KafkaStreamSrcConn

These parameters control the connection to the Kafka broker, registered on the collection via --conn-file (see the example above).

Parameter	Description
servers	Required. Bootstrap servers as a single comma-separated string of host[:port] items.
credentials	Optional. UserPassword for SASL/SCRAM auth; omit for an unauthenticated cluster.
group_id	Required. Consumer-group id used by the polling consumer.
schema_registry	Optional. ConfluentSchemaRegistry or AwsGlueSchemaRegistry; omit for schema-less ingest.
conn_cfg	Optional. Connector config forwarded to confluent_kafka.Consumer.

Schema Registry Configuration

If your Kafka messages rely on an external schema registry for deserialization (Avro or Protobuf), set schema_registry on the connection (KafkaStreamSrcConn), not on the function.

Confluent Schema Registry

Configuration object: tabsdatak.conn.kafka.ConfluentSchemaRegistry

Parameter	Type	Description
url	str	Required. The registry's HTTP(S) URL.
credentials	UserPassword	Optional. Basic-auth credentials.
configs	dict	Optional. Overrides forwarded to confluent_kafka.schema_registry.SchemaRegistryClient.

AWS Glue Schema Registry

Configuration object: tabsdatak.conn.kafka.AwsGlueSchemaRegistry

Parameter	Type	Description
region_name	str	Required. The AWS region the registry lives in, e.g. us-east-1.
registry_name	str	Required. The registry name in AWS Glue.
schema_name	str	Required. The schema name within the registry.
credentials	AwsAccessSecretKey	Required. Access key id and secret access key.
configs	dict	Optional. Overrides forwarded to the boto3 client.

Lifecycle Management (CLI)

You can pause or resume ingestion without deleting the publisher, using the same enable/disable flags as any other function:

# Stop a running streaming publisher
tdk fn disable --coll <COLLECTION_NAME> --name <FUNCTION_NAME>

# Resume it
tdk fn enable --coll <COLLECTION_NAME> --name <FUNCTION_NAME>

To register a publisher without immediately consuming data, pass --disabled at registration time, then enable it later with the command above.