> For clean Markdown of any page, append .md to the page URL. > For a complete documentation index, see https://ocient.ferndocs.com/guides/lat-reference/lat-pipeline-configuration/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) > Set up and manage data pipelines in Ocient for seamless data processing and integration.