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

# Data Pipeline Load of JSON Data from Kafka

export const TimeKey = "TimeKey®";

export const Ocient = "Ocient®";

export const Metabase = "Metabase℠";

export const Kafka = "Apache® Kafka®";

A common setup for streaming data into {Ocient} is to load JSON data from an {Kafka} topic.

Ocient uses data pipelines to transform each document into rows in one or more different tables. The loading and transformation capabilities use a simple SQL-like syntax for transforming data. This tutorial guides you through a simple example load using a small data set in JSON format. The data in this example comes from a test set for the {Metabase} Business Intelligence tool.

## Prerequisites

This tutorial assumes that:

1. The Ocient System has network access to a Kafka broker from the SQL and Loader Nodes.
2. An Ocient System is installed and configured with an active Storage Cluster (see the [Ocient Application Configuration](/ocient-application-configuration) guide).

## Step 1: Create a New Database

Connect to a SQL Node using the [Commands Supported by the Ocient JDBC CLI Program](/commands-supported-by-the-ocient-jdbc-cli-program). Then, execute the `CREATE DATABASE` SQL statement to create the `metabase` database.

```sql SQL theme={null}
CREATE DATABASE metabase;
```

## Step 2: Create Table

Create the `orders` table in the new database. First, connect to that database (e.g., `connect to jdbc:ocient://sql-node:4050/metabase`), and then execute this `CREATE TABLE` SQL statement that specifies to create a table with these columns and a clustering index based on the `user_id` and `product_id` columns:

* `created_at` as a {TimeKey} timestamp that is not nullable.
* `id`, `user_id`, and `product_id` as integers that are not nullable.
* `subtotal`, `tax`, `total`, and `discount` as floating point numbers.
* `quantity` as an integer.

```sql SQL theme={null}
CREATE TABLE public.orders (
    created_at TIMESTAMP TIME KEY BUCKET(30, 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)
);
```

The database creates the `orders` table, and you can begin loading data.

## Step 3: Create a Data Pipeline

Create data pipelines using the `CREATE PIPELINE` SQL statement. To load data, you first create a pipeline with the definition of the source, data format, and transformation rules using a SQL-like declarative syntax. Then, you execute the `START PIPELINE` command to start the load. You can observe progress and status using system tables and views.

Each Ocient pipeline defines a single data source and the target table or tables into which data loads. A data source includes the location of the source and filters on the source to define the specific data set to load. This example loads data from two data sources, where each source is located in a directory on the same S3 bucket.

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

```json JSON theme={null}
{"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}

```

In this case, Ocient automatically transforms the data to the target columns using some sensible conventions. In other cases, loads require some transformation. Most transformations are identical to functions that already exist in the SQL dialect of the Ocient System.

Create a pipeline named `orders_pipeline` for the orders data set from your database connection prompt. Specify the Kafka source with broker address `192.168.0.1:9092` (replace this example Kafka broker address with your address) and topic `orders`. Load the data into the `public.orders` table. The `SELECT` part of the SQL statement maps the JSON fields to the target columns in the created table.

<CodeGroup>
  ```sql SQL theme={null}
  CREATE PIPELINE orders_pipeline
  SOURCE
      KAFKA
          BOOTSTRAP_SERVERS '192.168.0.1:9092'
          TOPIC 'orders'
  EXTRACT
      FORMAT json
  INTO public.orders
  SELECT
      $id as id,
      $user_id as user_id,
      $product_id as product_id,
      $subtotal as subtotal,
      $tax as tax,
      $total as total,
      $discount as discount,
      $created_at as created_at,
      $quantity as quantity;
  ```
</CodeGroup>

The pipeline has three main sections:

* `SOURCE` — Loads data from Kafka. Specify the address of the bootstrap servers and topic. Options exist to set Kafka consumer configurations.
* `EXTRACT` — Sets the format to JSON.
* `INTO ... SELECT` — Targets the `public.orders` table and selects the chosen fields from the JSON records. In this case, all data is available at the top level of the JSON records, so the example references the fields by the attribute name (e.g., `$id`, `$user_id`, etc.). For nested data, reference the nested fields using dot notation (e.g., `$order.user.first_name`). Each field maps to a target column using the `as` syntax.

After you successfully create this pipeline, execute the `START PIPELINE` SQL statement.

<CodeGroup>
  ```sql SQL theme={null}
  START PIPELINE orders_pipeline;
  ```
</CodeGroup>

## Step 4: Confirm that Loading is Operating Correctly

With the pipeline in place and running, data immediately begins loading off of the Kafka topics that are configured in the pipeline. If you do not have data in the Kafka topics yet, now is a good time to start producing data into the topics.

### Produce Test Data into Kafka

For test purposes, [kafkacat](https://github.com/edenhill/kcat) is a helpful utility that makes it easy to produce records into a topic. For example, if you have a file of sample data `orders.jsonl` in a JSONL format (newline-delimited JSON records), you can run this command to send those records into your Kafka broker. Specify your broker IP address instead of `<broker_ip_address>` and your topic name `<topic_name>`.

```shell Shell theme={null}
kafkacat -b <broker_ip_address>:9092 -t <topic_name> -T -P -l orders.jsonl
```

Save an example document to your file system to use for this test. For this example, 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`.

Assuming the broker is running at the IP address `10.0.0.3`, send data into the `orders` topic defined in the pipeline definition by executing this command.

```shell Shell theme={null}
kafkacat -b 10.0.0.3:9092 -t orders -T -P -l orders.jsonl
```

This command pushes the entire JSONL file of messages into Kafka with one record per line. As these records are produced into Kafka, the running pipeline begins to load them into Ocient.

## Step 5: Observe the Load Progress

With your pipeline running, data immediately begins to load from the Kafka topic. The pipeline creates parallel Kafka consumers for each partition. If there are more partitions than processing cores available, the pipeline automatically handles spreading consumers across available processing cores and Loader Nodes.

You can check the pipeline status and progress by querying `information_schema.pipeline_status` or executing the `SHOW PIPELINE_STATUS` SQL statement.

<CodeGroup>
  ```sql SQL theme={null}
  SHOW PIPELINE_STATUS;
  ```
</CodeGroup>

*Output*

<CodeGroup>
  ```sql SQL theme={null}
  database_name         pipeline_name       table_names          status      status_message                               percent_complete duration_seconds files_processed     files_failed        files_remaining     records_processed   records_loaded      records_failed
  --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
  metabase              orders_pipeline     ["public.orders"]    RUNNING     Started processing pipeline orders_pipeline  0.0              2.025824         0                   0                   1                   0                   0                   0
  ```
</CodeGroup>

After a few seconds, the data is available for query in `public.orders`.

<CodeGroup>
  ```sql SQL theme={null}
  SELECT COUNT(*) FROM public.orders;
  ```
</CodeGroup>

*Output*

<CodeGroup>
  ```sql SQL theme={null}
  count(*)
  --------------------
  18760
  ```
</CodeGroup>

Unlike a batch file load, Kafka pipelines run continuously, so they never change to a status of `COMPLETED`. To examine progress, you can use the information schema and system catalog tables. Key details are in `information_schema.pipelines`, `information_schema.pipeline_status`, and `sys.pipeline_partitions`.

For example, this statement shows the status and key metrics for duration, loaded records, and failed records.

<CodeGroup>
  ```sql SQL theme={null}
  SELECT
      pipeline_name,
      status,
      status_message,
      duration_seconds,
      records_loaded,
      records_failed
  FROM
      information_schema.pipeline_status;
  ```
</CodeGroup>

*Output*

<CodeGroup>
  ```sql SQL theme={null}
  pipeline_name    status   status_message                               duration_seconds  records_loaded         records_failed
  --------------------------------------------------------------------------------------------------------------------------------
  orders_pipeline  RUNNING  Started processing pipeline orders_pipeline  10.089936         18760                  0
  ```
</CodeGroup>

In this example, there are no errors. The `sys.pipeline_errors` system catalog table captures any errors that occur during the pipeline process.

You can drop the pipeline with the `DROP PIPELINE orders_pipeline;` SQL statement. Execution of this statement leaves the data in your target table, but removes metadata about the pipeline execution from the system.

## Related Links

[Data Pipelines Reference](/data-pipelines)

[Load JSON Data](/data-formats-for-data-pipelines#load-json-data)

[Data Types for Data Pipelines](/data-types-for-data-pipelines)

[Transform Data in Data Pipelines](/transform-data-in-data-pipelines)

[Manage Errors in Data Pipelines](/manage-errors-in-data-pipelines)