---
title: LAT Load CSV Data from S3
slug: LCwxVhKnuDdvrxoLjtQG_
description: Learn to load CSV data from Amazon S3 directly into Ocient’s high-performance data warehouse, ensuring smooth and scalable data integration.
docTags: 
createdAt: Thu Jul 21 2022 17:05:13 GMT+0000 (Coordinated Universal Time)
updatedAt: Fri Apr 10 2026 21:08:33 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 common setup for batch loading files into Ocient is to load from a bucket on Amazon® Web Services℠ (AWS℠) S3 with time partitioned data. In many instances, a batch load is performed on a recurring basis to load new files. The LAT transforms each document into rows in one or more different tables. Ocient’s Loading and Transformation capabilities use a simple SQL-like syntax for transforming data. This tutorial will guide users through a simple example load using a small set of data in CSV format. The data in this example is created from a test set for the Metabase℠ Business Intelligence tool.

## Prerequisites

This tutorial assumes that:

1. The Ocient System has network access to S3 from the Loader Nodes.
2. An Ocient System is installed and configured with an active storage cluster (See the [Ocient Application Configuration](<././././System Administration/Install an Ocient System/Ocient Application Configuration.md>) guide).
3. Loading and Transformation is installed on the Loader Nodes.
4. A default "sink" for the Ocient Loader Nodes is configured on the system.
5. The [LAT Client Command Line Interface](<./././LAT Client Command Line Interface.md>) is installed.

## Step 1: Create a New Database

To begin, load two example tables in a database. First, connect to a SQL Node using the [Commands Supported by the Ocient JDBC CLI Program](<././././Connection Driver Reference/JDBC Manual/Commands Supported by the Ocient JDBC CLI Program.md>). Then run the following DDL command:

```sql
CREATE DATABASE metabase;
```

## Step 2: Create Tables

To create tables in the new database, first connect to that database (e.g., `connect to jdbc:ocient://sql-node:4050/metabase`), then run the following DDL commands:

```sql
CREATE TABLE public.orders (
  created_at TIMESTAMP TIME KEY BUCKET(1, DAY) NOT NULL,
  id INT NOT NULL,
  user_id INT NOT NULL,
  product_id INT NOT NULL,
  subtotal DOUBLE,
  tax DOUBLE,
  total DOUBLE,
  discount DOUBLE,
  quantity INT,
  CLUSTERING INDEX idx01 (user_id, product_id)
);

CREATE TABLE public.products(
  created_at TIMESTAMP TIME KEY BUCKET(1, DAY) NOT NULL,
  id INT NOT NULL,
  ean VARCHAR(255),
  title VARCHAR(255),
  category VARCHAR(255) COMPRESSION GDC(2) NOT NULL,
  vendor VARCHAR(255),
  price DOUBLE,
  rating DOUBLE,
  CLUSTERING INDEX idx01 (category)
);
```

Now, the database tables are created and you can begin loading data.

## Step 3: Create a Data Pipeline

Data pipelines are created using a simple loading configuration that is submitted to the Transformation Nodes to start loading. File Groups designate a batch of files to load. Each File Group is routed to one or more Ocient tables, and each column is the result of a transformation applied to the source document.

First, inspect the data that you plan to load. Each document has a format similar to the following example:

```Text
/* orders */
id,user_id,product_id,subtotal,tax,total,discount,created_at,quantity
1,1,14,37.65,2.07,39.72,null,2019-02-11T21:40:27.892Z,2
2,1,123,110.93,6.1,117.03,null,2018-05-15T08:04:04.580Z,3
3,1,105,52.72,2.9,49.2,6.42,2019-12-06T22:22:48.544Z,2
...

/* products */
id,ean,title,category,vendor,price,rating,created_at
1,1018947080336,Rustic Paper Wallet,Gizmo,"Swaniawski, Casper and Hilll",29.46,4.6,2017-07-19T19:44:56.582Z
2,7663515285824,Small Marble Shoes,Doohickey,Balistreri-Ankunding,70.08,0,2019-04-11T08:49:35.932Z
3,4966277046676,Synergistic Granite Chair,Doohickey,"Murray, Watsica and Wunsch",35.39,4,2018-09-08T22:03:20.239Z
...
```

This is similar to the target schema created in Step 2, but it will require some transformation. Most transformations are identical to functions already in Ocient’s SQL dialect. To route data to the tables, you need to create a pipeline.json file that has the following structure:

```json
{
    "version": 2,
    "workers": 4,
    "source": {
        "compression": "none",
        "type": "s3",
        "endpoint": "https://s3.us-east-1.amazonaws.com",
        "bucket": "ocient-docs",
        "file_groups": {
            "orders": {
                "prefix": "metabase_samples/csv",
                "file_matcher_syntax": "glob",
                "file_matcher_pattern": "**orders*.csv",
                "sort_type": "lexicographic"
            },
            "products": {
                "prefix": "metabase_samples/csv",
                "file_matcher_syntax": "glob",
                "file_matcher_pattern": "**products*.csv",
                "sort_type": "lexicographic"
            }
        }
    },
    "extract": {
        "orders": {
            "record_type": "delimited",
            "headers": [
                "id",
                "user_id",
                "product_id",
                "subtotal",
                "tax",
                "total",
                "discount",
                "created_at",
                "quantity"
            ],
            "null_strings": [
                "NULL",
                "N/A"
            ],
            "record_delimiter": "\r\n",
            "num_header_lines": 1
        },
        "products": {
            "record_type": "delimited",
            "headers": [
                "id",
                "ean",
                "title",
                "category",
                "vendor",
                "price",
                "rating",
                "created_at"
            ],
            "record_delimiter": "\r\n",
            "num_header_lines": 1
        }
    },
    "transform": {
        "file_groups": {
            "orders": {
                "tables": {
                    "metabase.public.orders": {
                        "columns": {
                            "id": "id",
                            "user_id": "user_id",
                            "product_id": "product_id",
                            "subtotal": "subtotal",
                            "tax": "tax",
                            "total": "total",
                            "discount": "discount",
                            "created_at": "to_timestamp(created_at, 'yyyy-MM-dd\\'T\\'HH:mm:ss[.SSS]X')",
                            "quantity": "quantity"
                        }
                    }
                }
            },
            "products": {
                "tables": {
                    "metabase.public.products": {
                        "columns": {
                            "id": "id",
                            "ean": "ean",
                            "title": "title",
                            "category": "category",
                            "vendor": "vendor",
                            "price": "price",
                            "rating": "rating",
                            "created_at": "to_timestamp(created_at, 'yyyy-MM-dd\\'T\\'HH:mm:ss[.SSS]X')"
                        }
                    }
                }
            }
        }
    }
}
```

The two interesting parts of this pipeline.json file are the way the file groups and the extraction settings are defined.&#x20;

First, note that each file group sets the S3 endpoint, a bucket, a prefix used for filtering the considered files, and then a file matcher. In this example, there is only a single file, but if there were many files matching the pattern `**orders*.csv` then they would all be part of the file group. You do not need S3 credentials because this is a public bucket, but if it were private, you can supply credentials in a few different ways.

Next, note the extract section. Here, the example specifies a delimited format. If it were compressed (e.g., gzip), you could specify compression. The example also specifies the headers to associate with each column in the CSV data. Delimited extracts can specify different record delimiters (e.g., `\n`), specify field delimiters (e.g., `|`, `\t`, `,`), define how to handle empty fields or to trim whitespace, and specify strings that should be considered `NULL`. While not used in this file, an example of null strings is provided that would turn the string literal "NULL" or "N/A" into a database `NULL`.

The final parameter supplied in the example is the sort type for the file load. This informs the LAT how you would like data to be ordered when loading. The ideal sort organizes files in time order according to the defined TimeKey®. This makes more efficient segments and is much faster to load. This example uses the lexicographic sort which orders according to the characters in the file name. Other sort types are available to use file modified time or to extract the timestamp for sorting from the file path or file name.

## Step 4: Using the Loading and Transformation CLI

With a pipeline.json file ready to go, you can test this pipeline. To test, use the LAT CLI. For these examples, assume that two LATs are configured and set using an environment variable.

First, configure the LAT CLI to use the hosts of the Ocient Loading and Transformation service. You can add these to every CLI command as a flag, but for simplicity you can also set them as environment variables. From a command line, run the following command replacing the IP addresses with the IP addresses of your LAT processes:

```shell
export LAT_HOSTS="https://10.0.0.1:8443,https://10.0.0.2:8443"
```

:::hint{type="info"}
If your LAT is running without TLS configured, replace the port number of your LAT Hosts with 8080 and the protocol with `http://`.
:::

Next, check on the status of the LAT:

```shell
lat_client pipeline status
```

**Example response:**

```bash
10.0.0.1:8443: Stopped
10.0.0.2:8443: Stopped
```

Success! This confirms that you can reach the LAT from your CLI. If the status is "Running" it means a pipeline is already executing a pipeline. In the next step, you will update and start the new pipeline.

This example uses secure connections. If you receive an SSL Error when testing, your service cannot be configured to use TLS or you might need to use the `--no-verify` flag if certificate validation fails.

## Step 5: Test the Transformation

The CLI supports previewing a transformation with an example document and the pipeline file. This makes it easy to test your transformations.

First, save an example document to your file system to use for this test. For this demo, you can download an example file from [https://ocient-docs.s3.amazonaws.com/metabase\_samples/csv/orders.csv](https://ocient-docs.s3.amazonaws.com/metabase_samples/csv/orders.csv) and save it to `~/orders.csv`.

Next, make sure the pipeline.json file that you created is stored at `~/pipeline.json`.

Now that both files are available, you can run the CLI to preview the results. Pass the preview command the topic name, the pipeline file, and the sample record file. The response contains the transformed data tied to the destination table and a list of any error records.

Similar to how you can preview records on a Apache® Kafka® topic for file loads, you can supply any one `file_groups` created in the extract section to preview the transformations.

```shell
lat_client preview --topic orders --pipeline ~/pipeline.json --records ~/orders.csv
```

**Example response:**

```json
{
    "tableRecords": {
        "metabase.public.orders": [
            {
                "id": 1,
                "user_id": 1,
                "product_id": 14,
                "subtotal": 37.65,
                "tax": 2.07,
                "total": 39.72,
                "discount": null,
                "created_at": 1549921227892000000,
                "quantity": 2
            },
            {
                "id": 2,
                "user_id": 1,
                "product_id": 123,
                "subtotal": 110.93,
                "tax": 6.1,
                "total": 117.03,
                "discount": null,
                "created_at": 1526371444580000000,
                "quantity": 3
            },
            {
                "id": 3,
                "user_id": 1,
                "product_id": 105,
                "subtotal": 52.72,
                "tax": 2.9,
                "total": 49.2,
                "discount": 6.42,
                "created_at": 1575670968544000000,
                "quantity": 2
            }
        ]
    },
    "recordErrors": []
}
```

You can see that the data is transformed and the columns to which each transformed value will be mapped. If there are issues in the values, these will appear in the `recordErrors` object. You can quickly update the pipeline.json file and preview again. Now, you can inspect different documents to confirm that various states of data cleanliness like missing columns, null values, and special characters are well handled by your transformations.

## Step 6: Configure and Start the Data Pipeline

With a tested transformation, the next step is to setup and start the data pipeline.

First, configure the pipeline using the `pipeline create` command. This validates and creates the pipeline, but will not take effect until you start the pipeline:

```shell
lat_client pipeline create --pipeline ~/pipeline.json
```

**Example response:**

```bash
10.0.0.1:8443: Created
10.0.0.2:8443: Created
```

:::hint{type="info"}
In cases where there is an existing pipeline operating, it is necessary to stop the pipeline and remove the original pipeline before creating and starting the new pipeline.
:::

Now that the pipeline has been created on all LAT Nodes, you can start the LAT by running the `pipeline start` commands:

```shell
lat_client pipeline start
```

**Example responses:**

```bash
10.0.0.1:8443: Running
10.0.0.2:8443: Running
```

## Step 7: Confirm that Loading is Operating Correctly

With your pipeline in place and running, data will immediately begin loading from the S3 file groups that you defined. If there were many files per file group, the LAT would first sort the files, then partition them for the fastest loading based on the sorting criteria you provided.

### Observing Loading Progress

With the pipeline running, data immediately begins to load into Ocient. To observe this progress, you can use the `pipeline status` command from the LAT Client or monitor the LAT metrics endpoint of the Loader Nodes.

You can check the status with this command by using the `--list-files` flag to include a summary of the files included in the load.

```shell
lat_client pipeline status --list-files
```

**Example responses:**

```bash
10.0.0.1:8443: Running
10.0.0.2:8443: Running

Pipeline Files Processed: 0
Pipeline Error Count: 0
Pipeline Files Remaining: 2

orders
  Files Processed: 0
  Error Count: 0
  Files Remaining: 1

products
  Files Processed: 0
  Error Count: 0
  Files Remaining: 1


orders

Status        Filename
processing    metabase_samples/csv/orders.csv

In Process Files:
processing    metabase_samples/csv/orders.csv


products

Status        Filename
processing    metabase_samples/csv/products.csv

In Process Files:
processing    metabase_samples/csv/products.csv
```

You can monitor the LAT metrics endpoint manually at the command line. Or, you can use a tool like Grafana® to retrieve metrics. For this example, run the curl command against the endpoint and review the result. For details on metrics, see the [LAT Metrics](<./././LAT Metrics.md>) Documentation.

**Command:**

```curl
curl https://127.0.0.1:8443/v2/metrics/lat:type=pipeline
```

:::hint{type="info"}
If your LAT is running without TLS configured, replace the port number of your LAT Hosts with 8080 and the protocol with `http://`.
:::

**Example response:**

```json
{
  "request": {
  "mbean": "lat:type=pipeline",
  "type": "read"
  },
  "value": {
  "partitions": [
    {
        "offsets_durable": 1,
        "pushes_errors": 0,
        "pushes_attempts": 1,
        "rows_pushed": 1,
        "offsets_written": 18759,
        "records_buffered": 0,
        "records_errors_column": 0,
        "records_errors_deserialization": 0,
        "source_bytes_buffered": 0,
        "records_errors_transformation": 0,
        "offsets_processed": 18759,
        "partition": "table_orders-0",
        "records_filter_accepted": 1,
        "records_errors_row": 0,
        "records_filter_rejected": 0,
        "records_errors_generic": 0,
        "producer_send_attempts": 0,
        "offsets_pushed": 18759,
        "pushes_unacknowledged": 0,
        "invalid_state": 0,
        "bytes_pushed": 88,
        "records_errors_total": 0,
        "offsets_buffered": 18759,
        "complete": 0,
        "offsets_end": 1,
        "producer_send_errors": 0
    },
    {
    "offsets_durable": 1,
    "pushes_errors": 0,
    "pushes_attempts": 1,
    "rows_pushed": 1,
    "offsets_written": 199,
    "records_buffered": 0,
    "records_errors_column": 0,
    "records_errors_deserialization": 0,
    "source_bytes_buffered": 0,
    "records_errors_transformation": 0,
    "offsets_processed": 199,
    "partition": "table_products-0",
    "records_filter_accepted": 1,
    "records_errors_row": 0,
    "records_filter_rejected": 0,
    "records_errors_generic": 0,
    "producer_send_attempts": 0,
    "offsets_pushed": 199,
    "pushes_unacknowledged": 0,
    "invalid_state": 0,
    "bytes_pushed": 145,
    "records_errors_total": 0,
    "offsets_buffered": 199,
    "complete": 0,
    "offsets_end": 1,
    "producer_send_errors": 0
    }
  ],
  "paused": 1,
  "bytes_buffered": 0,
  "workers": 20
  },
  "timestamp": 1626970368,
  "status": 200,
}
```

### Check Row Counts in Tables

To confirm that you are seeing results in the target tables, you can also run some simple queries to check row counts. Depending on the streamloader role settings, the time for records to become queryable can vary from a few seconds to minutes:

**Example Queries:**

```sql
Ocient> SELECT count(*) FROM public.orders;
COUNT(*)
---------------------
18760
```

```sql
Ocient> SELECT count() FROM public.products; 
COUNT()
--------------------
200
```

Now you can explore the data in these four tables with any Ocient SQL queries.&#x20;

### Check Errors

In this example, all rows load successfully. However, a successful load does not always happen, and you can inspect errors using the LAT Client. Whenever the LAT process fails to parse a file correctly or fails to transform or load a record, the LAT process records an error. The LAT Client includes the `lat_client pipeline errors` command that reports the latest errors on the pipeline.

A full error log is available on the Loader Nodes. These logs report all bad records and the reason that the load fails.

:::hint{type="info"}
When you load a pipeline from Kafka, the load might route errors to an error topic on the Kafka broker instead of the logs. The LAT Client does not contain the errors sent to the error topic. You can inspect these errors with Kafka utilities instead.
:::

This LAT Client command displays a maximum of 100 error messages.

```bash
lat_client pipeline errors --max-errors 100 --only-error-messages

|--------------------------------------------------|
| exception_message                                |
|--------------------------------------------------|
| Column name: time1. Message: Failed to evaluate  |
| expression. Cause:                               |
| java.time.format.DateTimeParseException          |
|--------------------------------------------------|
| Column name: time1. Message: Failed to evaluate  |
| expression. Cause:                               |
| java.time.format.DateTimeParseException          |
|--------------------------------------------------|
```

The errors indicate that there is an issue parsing the `time1` column. Options exist on the `pipeline errors` command to return JSON and to restrict the response to specific components of the error detail that includes a reference to the source location of this record.

The following command returns JSON that is delimited with newline characters. You can pass the JSON output to `jq` or a file. The JSON includes the source topic or file group, the filename where the error occurred, the offset that indicates the line number or Kafka offset, and the exception message that aids in troubleshooting and identifying the incorrect record in the source data. You can use the `log_original_message` pipeline setting to provide direct access to the parsed source record for errors when appropriate.

```bash
lat_client pipeline errors --max-errors 100 --json

{"time": "2022-05-17T16:53:50.387386+00:00", "topic": "calcs", "partition": 0, "state": "TRANSFORMATION_ERROR", "exception_message": "Column name: time1. Message: Failed to evaluate expression. Cause: java.time.format.DateTimeParseException: Cannot parse time \"19:36:22\" with format string \"Value(HourOfDay,2)Offset(+HHmm,'Z')':'Value(MinuteOfHour,2)':'Value(SecondOfMinute,2)\"\njava.time.format.DateTimeParseException: Cannot parse time \"19:36:22\" with format string \"Value(HourOfDay,2)Offset(+HHmm,'Z')':'Value(MinuteOfHour,2)':'Value(SecondOfMinute,2)\"\nCannot parse time \"19:36:22\" with format string \"Value(HourOfDay,2)Offset(+HHmm,'Z')':'Value(MinuteOfHour,2)':'Value(SecondOfMinute,2)\"", "offset": 0, "record": null, "metadata": {"size": "3321", "filename": "calcs/csv/calcs_01.csv"}}
{"time": "2022-05-17T16:53:50.404684+00:00", "topic": "calcs", "partition": 0, "state": "TRANSFORMATION_ERROR", "exception_message": "Column name: time1. Message: Failed to evaluate expression. Cause: java.time.format.DateTimeParseException: Cannot parse time \"02:05:25\" with format string \"Value(HourOfDay,2)Offset(+HHmm,'Z')':'Value(MinuteOfHour,2)':'Value(SecondOfMinute,2)\"\njava.time.format.DateTimeParseException: Cannot parse time \"02:05:25\" with format string \"Value(HourOfDay,2)Offset(+HHmm,'Z')':'Value(MinuteOfHour,2)':'Value(SecondOfMinute,2)\"\nCannot parse time \"02:05:25\" with format string \"Value(HourOfDay,2)Offset(+HHmm,'Z')':'Value(MinuteOfHour,2)':'Value(SecondOfMinute,2)\"", "offset": 1, "record": null, "metadata": {"size": "3321", "filename": "calcs/csv/calcs_01.csv"}}
```

## Related Links

[LAT Overview](<./././LAT Overview.md>)

[LAT Data Types in Loading](<./././LAT Data Types in Loading.md>)

[LAT Advanced Topics](<./././LAT Advanced Topics.md>)
