Sources

How-to guideApplies to: Unified engine deployments

A source defines a connection and the columns a pipeline can read. This page uses Kafka with JSON records. Complete Before you start before registering it.

Register a Kafka source

If you already ran the walkthrough, orders_source is registered. Reuse it in transformations. Otherwise, edit the connection values and run this declaration once.

CREATE TABLE orders_source (
  order_id BIGINT,
  customer_id BIGINT,
  amount DOUBLE,
  status TEXT
) WITH (
  connector = 'kafka',
  type = 'source',
  bootstrap_servers = 'kafka:9092',
  topic = 'orders',
  format = 'json',
  'source.offset' = 'earliest',
  'source.read_mode' = 'read_committed'
);

The result contains a table_id and accepted: true. This saves the connector definition. Records are consumed when a pipeline using it starts.

Match the message schema

The declaration above reads JSON objects like this one:

{"order_id":1,"customer_id":101,"amount":20,"status":"paid"}

Field names and value types must match the declared columns. BIGINT holds whole-number IDs, DOUBLE holds amounts, and TEXT holds strings. The walkthrough uses JSON objects with fixed columns. Avro and Protobuf need their own matching format and schema settings.

Choose offsets and read mode

Option	Values and behavior	Default
`source.offset`	`earliest` starts at the earliest retained records; `latest` starts at the end; `group` uses stored consumer group offsets	`latest`
`source.read_mode`	`read_committed` reads committed transactional records; `read_uncommitted` also allows uncommitted records	`read_uncommitted`
`source.group_id`	An explicit Kafka consumer group ID	Generated for the job and source operator
`source.group_id_prefix`	Prefix for a generated group ID; an explicit group ID takes priority	Unset
`client_configs`	Additional Kafka client settings as comma-separated `key=value` pairs	Unset

The offset option chooses a starting position when no saved Streaming checkpoint is being restored. Resuming the same pipeline uses its checkpointed offsets. Creating a different pipeline with earliest can read the retained records again. For group, use the explicit group ID whose stored offsets you intend to read. Share a consumer group only when that sharing is part of your Kafka design.

Connect with SASL

SASL is Kafka’s authentication mechanism. This alternative uses a secured broker and worker environment variables for its username and password. It registers a separate source; use this source name in your pipeline query.

CREATE TABLE secured_orders_source (
  order_id BIGINT,
  customer_id BIGINT,
  amount DOUBLE,
  status TEXT
) WITH (
  connector = 'kafka',
  type = 'source',
  bootstrap_servers = 'secure-kafka:9093',
  topic = 'orders',
  format = 'json',
  'source.offset' = 'earliest',
  'source.read_mode' = 'read_committed',
  'auth.type' = 'sasl',
  'auth.protocol' = 'SASL_SSL',
  'auth.mechanism' = 'PLAIN',
  'auth.username' = '{{ KAFKA_USERNAME }}',
  'auth.password' = '{{ KAFKA_PASSWORD }}'
);

Choose the protocol and mechanism accepted by your broker. The {{ KAFKA_USERNAME }} and {{ KAFKA_PASSWORD }} references use the Streaming connector’s environment expansion. Supply those variables to the Streaming processes that resolve the connection, including its workers. They are separate from Unified secret or credential names. For AWS MSK IAM authentication, use the following option fragment inside a source declaration. It uses the runtime’s AWS identity, with an optional assumed role.

'auth.type' = 'aws_msk_iam',
'auth.region' = 'us-east-1',
'auth.role_arn' = 'arn:aws:iam::123456789012:role/streaming-kafka-reader'

Use the source in a pipeline

The FROM clause names the registered source. The target sink must also be registered. Your first pipeline shows the complete script. In the current Kafka connector, type must be source. Use an Iceberg sink for the destination shown in these examples.

Clean up

Stop and remove pipelines that use the source before removing its connector definition through the existing table controls. Removing a connector definition does not delete its Kafka topic.

Transformation Engine

Learn Laminar

Workloads

SQL reference

HTTP API

Recipes

Operations

Register a Kafka source

Match the message schema

Choose offsets and read mode

Connect with SASL

Use the source in a pipeline

Clean up

​Register a Kafka source

​Match the message schema

​Choose offsets and read mode

​Connect with SASL

​Use the source in a pipeline

​Clean up

Register a Kafka source

Match the message schema

Choose offsets and read mode

Connect with SASL

Use the source in a pipeline

Clean up