---
title: LAT Pipeline Configuration
slug: CFOJ-BHPcmO9O8YPlKqjW
description: Set up and manage data pipelines in Ocient for seamless data processing and integration.
docTags: 
createdAt: Thu Jul 21 2022 17:05:13 GMT+0000 (Coordinated Universal Time)
updatedAt: Fri Apr 10 2026 20:36:18 GMT+0000 (Coordinated Universal Time)
---

:::hint{type="warning"}
Data Pipelines are now the preferred method for loading data into the Ocient® System. For details, see [Load Data](<././Load Data.md>).
:::

A pipeline configuration is a JSON file that fully describes the necessary elements to run a pipeline, including

- **Source** - the source location from which a pipeline should read records to process
- **Extract** - how a pipeline should extract data from the source
- **Transform** - how a pipeline should transform incoming records
- **Sink** - the destination where a pipeline should write transformed rows

The JSON file is a list of keys with (possibly nested) values. The available key-value pairs are documented in these sections.

## Pipeline

Top level configuration for a pipeline.

Required keys:

- [version](./#version)&#x20;
- [source](./#source)&#x20;
- [extract](./#extract)
- [transform](./#transform) &#x20;

The following is an example of the structure of a pipeline configuration.

```json
{
    "version": 2,
    "source": {
        "type": "kafka"
        // kafka source configuration
    },
    "sink": {
        "type": "ocient"
        // ocient sink configuration
    },
    "extract": {
        // extract configuration
    },
    "transform": {
        "topics": {
            "my-topic": {
                "tables": {
                    "my-table": {
                        "columns": {
                            "my-col1": "record_field_1",
                            "my-col2": "record_field_2"
                        }
                    }
                }
            }
        }
    }
}
```

## Configuration

### `version`

The pipeline’s version. Required value is `2`.

| Type:     | `int` |
| --------- | ----- |
| Required: | Yes   |
| Default:  |       |

### `pipeline_id`

A unique identifier for this pipeline. Allowed characters are `a-z`, `A-Z`, `0-9`, `_`, and `-`.

The `pipeline_id` is used to uniquely identify a pipeline. It is used for a few purposes:

Deduplication scope. See [Understanding Deduplication](<./LAT Advanced Topics.md>).

1. For Apache® Kafka® loads, the consumer `group.id` is set to `ocient-lat-[pipeline_id]`.

:::hint{type="info"}
For most loads from File Sources, it is advisable to leave the `pipeline_id` unset when creating a pipeline using the LAT client. The client will assign the pipeline a random UUID.
:::

| Type:     | `string`                            |
| --------- | ----------------------------------- |
| Required: | No                                  |
| Default:  | LAT client randomly generated UUID. |

### `workers`

The number of workers this pipeline should use for processing records.

| Type:     | `int`                                          |
| --------- | ---------------------------------------------- |
| Required: | No                                             |
| Default:  | `DEFAULT_NUM_WORKERS` in service configuration |

### `log_original_records`

Add original records to the error log when errors occur. When this setting is set to `true`, LAT writes data extracted from the source to the error log and in some error messages. By default, this setting is `false` and source data is not written to the error log nor included in error messages.

In order to enable this setting, the [LAT\_ALLOW\_LOG\_ORIGINAL\_RECORDS](<./LAT Service Configuration.md>) service configuration must also be enabled.

:::hint{type="info"}
This configuration only affects pipelines that do not use an [error\_topic](./#errortopic).
:::

### `seek_on_rebalance`

Whether to seek a newly assigned partition to the latest known durable record prior to resuming processing. Disabling this behavior should typically be reserved for test scenarios and is only supported for Kafka loading.

| Type:     | `boolean` |
| --------- | --------- |
| Required: | No        |
| Default:  | `true`    |

### `continue_on_unrecoverable_error`

Whether to allow workers to continue processing when they encounter an ordinarily unrecoverable error.

| Type:     | `boolean` |
| --------- | --------- |
| Required: | No        |
| Default:  | `false`   |

### `single_file_mode`

Enable the single file mode. This mode is designed for a specific use case where there are few files but each file size is large. When using this mode, only a single file is processed at a time, so [Common File Source Configuration](<./LAT Pipeline Configuration/LAT Source Configuration.md>) must be equal to `1` and only one [Common File Source Configuration](<./LAT Pipeline Configuration/LAT Source Configuration.md>) can be defined. The single file is processed in parallel by the number of workers defined by the pipeline `workers` setting. There is no need to enable this in common use cases.

| Type:     | `boolean` |
| --------- | --------- |
| Required: | No        |
| Default:  | `false`   |

:::hint{type="warning"}
A known limitation exists with the LAT metrics when you use the single file mode. Metrics returned from the `lat_client pipeline status` command might not display the expected count of files processed, processing, and so on. However, the load still displays the correct `PROCESSING` and `COMPLETED` statuses.
:::

### `error_topic`

A Kafka Topic to write records which cannot be processed. If absent, error records will be logged to the error log file without additional processing.

:::hint{type="info"}
This configuration is only available if a Kafka Source is configured for the pipeline. The configuration for that source will apply to the Kafka Producer for this topic.
:::

| Type:     | `string` |
| --------- | -------- |
| Required: | No       |
| Default:  | `null`   |

### `polling_duration`

Maximum duration to block while polling for new records from a Source, in milliseconds.

| Type:     | `int`  |
| --------- | ------ |
| Required: | No     |
| Default:  | `1000` |

### `source`

Source configuration section. See [LAT Source Configuration](<./LAT Pipeline Configuration/LAT Source Configuration.md>) for nested configuration.

| Type:     | `object` |
| --------- | -------- |
| Required: | Yes      |
| Default:  |          |

### `sink`

Sink configuration section. See [LAT Sink Configuration](<./LAT Pipeline Configuration/LAT Sink Configuration.md>) for inline configuration details. `sink` cannot be set if `sink_name` is set. `sink` can be omitted if a default sink is defined as an [External Sink Configuration](<./LAT Pipeline Configuration/LAT Sink Configuration.md>).

| Type:     | `object` |
| --------- | -------- |
| Required: | No       |
| Default:  | `null`   |

### `sink_name`

The name of an externally configured sink. `sink_name` cannot be set if [sink](./#sink)  is set. See [External Sink Configuration](<./LAT Pipeline Configuration/LAT Sink Configuration.md>). `sink_name` can be omitted if a default sink is defined as an [External Sink Configuration](<./LAT Pipeline Configuration/LAT Sink Configuration.md>).

| Type:     | `string` |
| --------- | -------- |
| Required: | No       |
| Default:  | `null`   |

### `extract`

Extract configuration section. See [LAT Extract Configuration](<./LAT Pipeline Configuration/LAT Extract Configuration.md>) for nested configuration.

| Type:     | `object`                                           |
| --------- | -------------------------------------------------- |
| Required: | No                                                 |
| Default:  | Defaults to JSON record type with default settings |

### `transform`

Transform configuration section. See [LAT Transform Configuration](<./LAT Pipeline Configuration/LAT Transform Configuration.md>) for nested configuration.

| Type:     | `object` |
| --------- | -------- |
| Required: | Yes      |
| Default:  |          |

## Related Links

[Ingest Data with Legacy LAT Reference](<./Ingest Data with Legacy LAT Reference.md>)
