> ## Documentation Index
> Fetch the complete documentation index at: https://docs.ocient.com/llms.txt
> Use this file to discover all available pages before exploring further.

# LAT Load JSON Data from S3

export const TimeKey = "TimeKey®";

export const Ocient = "Ocient®";

export const Metabase = "Metabase℠";

export const Kafka = "Apache® Kafka®";

export const AWS = "Amazon® Web Services℠ (AWS℠)";

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

A common setup for batch loading files into Ocient is to load from a bucket on {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 JSONL (newline delimited JSON) 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](/ocient-application-configuration) 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) is installed.

## Step 1: Create a New Database

To begin, you are going to load two example tables in a database. First, connect to a SQL Node using the [Commands Supported by the Ocient JDBC CLI Program](/commands-supported-by-the-ocient-jdbc-cli-program). Then run the following DDL command:

```sql SQL theme={null}
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 SQL theme={null}
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, let’s inspect the data you plan to load. Each document has a format similar to the following example:

```json JSON theme={null}
/* orders */
{"id": 1, "user_id": 1, "product_id": 14, "subtotal": 37.65, "tax": 2.07, "total": 39.72, "discount": null, "created_at": "2019-02-11T21:40:27.892Z", "quantity": 2}
{"id": 2, "user_id": 1, "product_id": 123, "subtotal": 110.93, "tax": 6.1, "total": 117.03, "discount": null, "created_at": "2018-05-15T08:04:04.580Z", "quantity": 3}
...

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

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

```json JSON theme={null}
{
    "version": 2,
    "workers": 4,
    "source": {
        "type": "s3",
        "endpoint": "https://s3.us-east-1.amazonaws.com",
        "bucket": "ocient-docs",
        "compression": "none",
        "file_groups": {
            "orders": {
                "prefix": "metabase_samples/jsonl",
                "file_matcher_syntax": "glob",
                "file_matcher_pattern": "**orders*.jsonl",
                "sort_type": "lexicographic"
            },
            "products": {
                "prefix": "metabase_samples/jsonl",
                "file_matcher_syntax": "glob",
                "file_matcher_pattern": "**products*.jsonl",
                "sort_type": "lexicographic"
            }
        }
    },
    "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 most interesting part of this pipeline.json file is the way it defines the file groups. Note that each sets the S3 endpoint, a bucket, a prefix used for filtering the considered files, and then a file matcher. In this case you only have a single file, but if there were many files matching the pattern `**orders*.jsonl` then they would all be part of the file group.

The final parameter that you supplied 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 is in time order according to the defined {TimeKey}. This makes more efficient segments and is much faster to load. In this case, you used the lexicographic sort which orders according to the characters in the filename. Other sort types are available to use file modified time or to extract the timestamp for sorting from the file path or filename.

## 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 Shell theme={null}
export LAT_HOSTS="https://10.0.0.1:8443,https://10.0.0.2:8443"
```

<Info>
  If your LAT is running without TLS configured, replace the port number of your LAT Hosts with 8080 and the protocol with `http://`.
</Info>

Next, check on the status of the LAT:

```shell Shell theme={null}
lat_client pipeline status
```

**Example response:**

```bash Bash theme={null}
10.0.0.1:8443: Stopped
10.0.0.2:8443: Stopped
```

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. Next, you are going to update and start your new pipeline.

This example uses secure connections. If you receive an SSL Error when testing, your service might not be configured to use TLS or you might need to use the `--no-verify` flag if the 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/jsonl/orders.jsonl](https://ocient-docs.s3.amazonaws.com/metabase_samples/jsonl/orders.jsonl) and save it to `~/orders.jsonl`.

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 {Kafka} topic for file loads, you can supply any one of the topics you created as file groups to preview the transformations.

```shell Shell theme={null}
lat_client preview --topic orders --pipeline ~/pipeline.json --records ~/orders.jsonl
```

**Example response:**

```json JSON theme={null}
{
    "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 your 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, you must 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 Shell theme={null}
lat_client pipeline create --pipeline ~/pipeline.json
```

**Example response:**

```bash Bash theme={null}
10.0.0.1:8443: Created
10.0.0.2:8443: Created
```

<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.
</Info>

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

```shell Shell theme={null}
lat_client pipeline start
```

Example responses:

```bash Bash theme={null}
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 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 Shell theme={null}
lat_client pipeline status --list-files
```

**Example responses:**

```bash Bash theme={null}
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/jsonl/orders.jsonl

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

products

Status        Filename
processing    metabase_samples/jsonl/products.jsonl

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

**Command:**

```curl CURL theme={null}
curl https://127.0.0.1:8443/v2/metrics/lat:type=pipeline
```

<Info>
  If your LAT is running without TLS configured, replace the port number of your LAT Hosts with 8080 and the protocol with `http://`.
</Info>

**Example response:**

```json JSON theme={null}
{
  "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 SQL theme={null}
Ocient> SELECT count(*) FROM public.orders;
COUNT(*)
----------------------
18760
```

```sql SQL theme={null}
Ocient> SELECT count() FROM public.products;
COUNT()
--------------------
200
```

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

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

<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.
</Info>

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

```bash theme={null}
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 theme={null}
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)

[LAT Data Types in Loading](/lat-data-types-in-loading)

[LAT Advanced Topics](/lat-advanced-topics)