> For the complete documentation index, see [llms.txt](https://v2.dataos.info/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://v2.dataos.info/references/v1/resources/nilus/stream/pipeline-config.md). # Stream pipeline config Nilus stream pipelines are authored as `type: nilus` resources with `spec.type: stream`. This page explains the current Nilus **stream** pipeline shape, long-running consumers that pull records from event-streaming systems (Kafka, NATS+JetStream) in micro-batches and write them to the configured sink. Stream pipelines do not run on a schedule; they stay up and consume continuously while `run_in_loop: true` is set. For sample configs by connector, see [Stream Sample Configs](/references/v1/resources/nilus/stream/sample-configs.md). For the connector-specific options, see the [Kafka](/references/v1/resources/nilus/stream/stream-sources/kafka.md) and [NATS](/references/v1/resources/nilus/stream/stream-sources/nats.md) pages.
Example Config YAML ```yaml name: nilus-orders-stream version: v1alpha type: nilus tags: - nilus-stream description: Stream Kafka orders into the lakehouse spec: type: stream compute: universe-compute logLevel: INFO resources: requests: cpu: "200m" memory: "256Mi" use: projection: secrets: - id: engineering:kafka-secret contextAlias: kafkasecret projections: envVars: - key: KAFKA_USERNAME template: "{{ secrets['kafkasecret'].username | base64_decode }}" - key: KAFKA_PASSWORD template: "{{ secrets['kafkasecret'].password | base64_decode }}" source: address: kafka://?bootstrap_servers=broker.prod:9093&group_id=nilus-prod&security_protocol=SASL_SSL&sasl_mechanisms=SCRAM-SHA-512&sasl_username={KAFKA_USERNAME}&sasl_password={KAFKA_PASSWORD}&batch_size=2000&batch_timeout=10 options: source_table: orders run_in_loop: true sink: address: dataos://orderslakehouse?purpose=rw options: dest_table: analytics.orders_stream incremental_strategy: append ```
## Configuration elements Fields are grouped below by function. ### 1. Pipeline metadata | Field | Description | | ------------- | ----------------------------------------------------- | | `name` | Unique pipeline identifier. | | `version` | Use `v1alpha` for the current Nilus resource shape. | | `type` | Must be `nilus` for Nilus-managed pipelines. | | `tags` | Optional labels for search, grouping, and operations. | | `description` | Optional human-readable summary. | ### 2. Nilus spec The `spec` block defines the stream pipeline contract that Nilus validates and renders into a long-running service. | Field | Required | Description | | ----------- | ------------------- | ----------------------------------------------------------------------- | | `spec.type` | Yes | Must be `stream` for streaming pipelines. | | `compute` | Yes | Compute profile used to run the long-running consumer. | | `logLevel` | No(default: `INFO`) | Optional log level: `DEBUG`, `INFO`, `WARNING`, or `ERROR`. | | `runAsUser` | No | Optional runtime identity. When omitted, Nilus uses the resource owner. | | `resources` | No | Optional CPU and memory requests or limits. | | `use` | No | Optional secret projection rules for direct-URI credentials. | | `source` | Yes | Defines how Nilus connects to the streaming system. | | `sink` | Yes | Defines where Nilus writes the consumed records. | {% hint style="info" %} Stream pipelines are long-running services. Do **not** add a `schedule` block, there is nothing to trigger on a cron because the consumer stays connected. Use Nilus Manager to start, stop, and observe a stream service. {% endhint %} ### 3. Source The `source` block defines how Nilus connects to the streaming system. ```yaml spec: source: address: kafka://?bootstrap_servers=broker.prod:9093&group_id=nilus-prod&batch_size=2000&batch_timeout=10 options: source_table: orders run_in_loop: true ``` * `address` is a direct connector URI (`kafka://`, `nats+jetstream://`) or a `dataos://` depot reference for clusters managed centrally in DataOS. * Connection options that govern transport (brokers, host/port, security protocol, SASL/SSL material, NATS subject) live as **query parameters** on the URI. * `options.source_table` is the topic (Kafka) or JetStream stream name (NATS). * `options.run_in_loop` keeps the consumer connected. Without it, Nilus exits after the first pull, which is rarely what you want for stream pipelines. #### `source.options` | Option | Required | What it does | Typical shape | | ------------------- | ----------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------ | | `source_table` | Yes | Logical source object. Topic for Kafka, JetStream stream name for NATS. | `orders` | | `run_in_loop` | Recommended | Keep the consumer running and continue pulling new batches. Set `true` for production stream services. | `true` | | `primary_key` | No | Logical row identifier used mainly with merge-style sinks. | `order_id` | | `mask` | No | Column masking rules. Object whose values are algorithm strings (`hash`, `redact`, `partial:3`). | `{ email: hash }` | | `max_table_nesting` | No | Schema hint that caps how deeply nested fields are flattened. Nodes deeper than this level are loaded as a struct or JSON column instead of being expanded into their own table. Useful for Kafka or NATS streams carrying nested JSON payloads. `0` keeps the runtime default. | `0` (default): increase to flatten nested fields | Connector-specific options, `bootstrap_servers`, `group_id`, `batch_size`, `batch_timeout`, `security_protocol`, `sasl_*`, `ssl_*` for Kafka; `subject`, `durable`, `batch_size`, `timeout`, `token`, `nkeys_seed` for NATS, go on the URI query string and are documented on the [Kafka](/references/v1/resources/nilus/stream/stream-sources/kafka.md) and [NATS](/references/v1/resources/nilus/stream/stream-sources/nats.md) source pages. #### Direct URI vs depot address * Use a direct connector URI (`kafka://...`, `nats+jetstream://...`) when you want to assemble the connection explicitly and project secrets through `spec.use.projection`. * Use `dataos://?purpose=rw` when the connection should come from a DataOS depot that already has credentials and TLS material configured. ### 4. Secrets and projections Stream sources almost always need credentials. For direct URIs, project secrets under `spec.use.projection` and reference them as `{ENV_VAR}` placeholders inside the URI: ```yaml spec: use: projection: secrets: - id: engineering:kafka-secret contextAlias: kafkasecret projections: envVars: - key: KAFKA_USERNAME template: "{{ secrets['kafkasecret'].username | base64_decode }}" - key: KAFKA_PASSWORD template: "{{ secrets['kafkasecret'].password | base64_decode }}" source: address: kafka://?bootstrap_servers=broker.prod:9093&group_id=nilus-prod&sasl_username={KAFKA_USERNAME}&sasl_password={KAFKA_PASSWORD} ``` When a `dataos://` depot is used, credentials and TLS material come from the depot, no projection is required. ### 5. Sink The `sink` block defines where Nilus writes the consumed records. Streaming pipelines almost always use `incremental_strategy: append` because every record represents a new event in the source. ```yaml spec: sink: address: dataos://orderslakehouse?purpose=rw options: dest_table: analytics.orders_stream incremental_strategy: append ``` #### `sink.options` | Option | Required | What it does | Typical shape | | ---------------------- | ---------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | -------------------------- | | `dest_table` | Yes | Target object in the destination. | `analytics.orders_stream` | | `incremental_strategy` | Yes | Write behavior across micro-batches. `append` is the default for stream sinks. Use `merge` only when the destination has a stable primary key and deduplication is genuinely needed. | `append` | | `primary_key` | When `incremental_strategy: merge` | Column(s) used to deduplicate. | `order_id` | | `partition_by` | No | Partitioning hint for lakehouse-style destinations. | structured partition rules | | `cluster_by` | No | Clustering hint for destinations that support it. | `customer_id` | | `loader_file_size` | No | Controls how many rows Nilus writes per output file or loader batch. | `100000` | Destination-specific knobs (region, catalog, warehouse, file format, etc.) live on the destination page rather than this stream-config reference. ### 6. Execution model Stream pipelines are continuously running workloads. The important authoring consequences: * Author the `nilus` resource in `v1alpha`. * Set `spec.type: stream` so Nilus treats the source as an ongoing event consumer. * Keep the operational intent in `spec`. * Stream services persist offsets/sequences in the source (Kafka consumer-group offsets, NATS durable consumers). A pipeline restart resumes from the last committed position **only** if the consumer identity (`group_id` for Kafka, `durable=` for NATS) is stable. ## Validation notes * Use `type: nilus` with `spec.type: stream` for streaming pipelines. Misclassifying a stream as `spec.type: batch` makes Nilus run the consumer once and exit. * Always set `run_in_loop: true` under `source.options`. * Do **not** add `spec.schedule`, stream services are continuously running, not cron-triggered. * Use `spec.use.projection` for direct URI credentials. Use `dataos://...?purpose=rw` when a depot should supply credentials automatically. * Keep `group_id` (Kafka) or `durable` (NATS) stable across deployments so offsets survive restarts. * Prefer `incremental_strategy: append` for stream sinks; reach for `merge` only when the destination has a primary key and deduplication is required. * Connector-specific source/sink options remain documented on the [Kafka](/references/v1/resources/nilus/stream/stream-sources/kafka.md) and [NATS](/references/v1/resources/nilus/stream/stream-sources/nats.md) pages and on each destination page. ## Related docs * [Understanding Stream Data Movement](/references/v1/resources/nilus/stream.md) * [Stream Sample Configs](/references/v1/resources/nilus/stream/sample-configs.md) * [Secrets and Projections](/references/v1/resources/nilus/concepts/secrets-and-projections.md) * [Understanding Batch Pipeline Config](/references/v1/resources/nilus/batch/pipeline-config.md) * [Understanding CDC Pipeline Config](/references/v1/resources/nilus/cdc/pipeline-config.md) * [Kafka](/references/v1/resources/nilus/stream/stream-sources/kafka.md) * [NATS](/references/v1/resources/nilus/stream/stream-sources/nats.md) --- # Agent Instructions This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com. ## Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter: ``` GET https://v2.dataos.info/references/v1/resources/nilus/stream/pipeline-config.md?ask=&goal= ``` `ask` is the immediate question: it should be specific, self-contained, and written in natural language. `goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.