> ## Documentation Index
> Fetch the complete documentation index at: https://docs.e6data.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Sources

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

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](https://docs.e6data.com/transformation/streaming/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.

```sql theme={"theme":{"light":"github-light","dark":"github-dark"}}
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:

```json theme={"theme":{"light":"github-light","dark":"github-dark"}}
{"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.

```sql theme={"theme":{"light":"github-light","dark":"github-dark"}}
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.

```text theme={"theme":{"light":"github-light","dark":"github-dark"}}
'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](https://docs.e6data.com/transformation/streaming/first-pipeline) shows the complete script.

In the current Kafka connector, `type` must be `source`. Use an [Iceberg sink](https://docs.e6data.com/transformation/streaming/sinks) 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.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.