> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://ocient.ferndocs.com/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://ocient.ferndocs.com/_mcp/server.

# LAT Pipeline Configuration

> Set up and manage data pipelines in Ocient for seamless data processing and integration.

> **Warning**
>
> Data Pipelines are now the preferred method for loading data into the Ocient System. For details, see [**Load Data**](/guides/load-data).

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)
* [**source**](#source)
* [**extract**](#extract)
* [**transform**](#transform)

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**](/guides/lat-reference/lat-advanced-topics).

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

> **Note**
>
> 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**](/guides/lat-reference/lat-service-configuration) service configuration must also be enabled.

> **Note**
>
> 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**](/guides/lat-reference/lat-pipeline-configuration/lat-source-configuration) must be equal to `1` and only one [**Common File Source Configuration**](/guides/lat-reference/lat-pipeline-configuration/lat-source-configuration) 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`   |

> **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.

> **Note**
>
> 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**](/guides/lat-reference/lat-pipeline-configuration/lat-source-configuration) for nested configuration.

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

### `sink`

Sink configuration section. See [**LAT Sink Configuration**](/guides/lat-reference/lat-pipeline-configuration/lat-sink-configuration) 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**](/guides/lat-reference/lat-pipeline-configuration/lat-sink-configuration).

| 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**](/guides/lat-reference/lat-pipeline-configuration/lat-sink-configuration). `sink_name` can be omitted if a default sink is defined as an [**External Sink Configuration**](/guides/lat-reference/lat-pipeline-configuration/lat-sink-configuration).

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

### `extract`

Extract configuration section. See [**LAT Extract Configuration**](/guides/lat-reference/lat-pipeline-configuration/lat-extract-configuration) for nested configuration.

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

### `transform`

Transform configuration section. See [**LAT Transform Configuration**](/guides/lat-reference/lat-pipeline-configuration/lat-transform-configuration) for nested configuration.

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

## Related Links

[**Ingest Data with Legacy LAT Reference**](/guides/lat-reference/ingest-data-with-legacy-lat-reference)