# Kafka Pipeline Using JSON Format

Kafka pipelines can use JSON formatted data for ingest. The minimum required syntax is below:

```sql
CREATE PIPELINE <pipeline_name> AS
LOAD DATA KAFKA '<host.example.com/my-topic>'
INTO TABLE <table_name>
FORMAT JSON;
```

To skip errors without skipping the entire batch, use the `SKIP PARSER ERRORS` clause.

```sql
CREATE PIPELINE <pipeline_name> AS
LOAD DATA KAFKA '<host.example.com/my-topic>'
INTO TABLE <table_name>
SKIP PARSER ERRORS 
FORMAT JSON;
```

When used with a stored procedure pipeline, `SKIP PARSER ERRORS` applies only to parsing errors that occur before the stored procedure is invoked. Constraint violations, duplicate key errors, and other data integrity issues must be handled within the procedure body.

```sql
CREATE PIPELINE <pipeline_name>
AS LOAD DATA KAFKA '<host.example.com/my-topic>'
INTO PROCEDURE <proc_name>
SKIP PARSER ERRORS
FORMAT JSON
(<json_column_mappings>);

```

> **📝 Note**: Skipped parser errors for pipelines are logged in the `information_schema.PIPELINES_ERRORS` table.

The `SKIP ALL ERRORS` clause can also be used to skip all erroneous messages. It is added in the `CREATE PIPELINE` statement. For example:

```sql
CREATE PIPELINE <pipeline_name> AS
LOAD DATA KAFKA '<host.example.com/my-topic>'
INTO TABLE <table_name>
SKIP ALL ERRORS 
FORMAT JSON;
```

Refer to [SKIP ALL ERRORS](https://docs.singlestore.com/db/v9.1/reference/sql-reference/data-manipulation-language-dml/load-data/#UUID-8d8456d6-b63b-4a11-94e1-da02f8ee4083.md) for more information.

> **❗ Important**: The use of `SKIP PARSER ERRORS` and `SKIP ALL ERRORS` along with `WITH TRANSFORM` is not supported. Using both clauses will cause an error when creating or altering a pipeline.

If the `SKIP PARSER ERRORS` clause is added during create pipeline, the `pipelines_stop_on_error` configuration will be ignored.

* The warning message will be logged in the `pipelines_errors` table, adding the Kafka message in the `LOAD_DATA_LINE` column.
* The entire Kafka message which has the parse error will be logged.
* Skip the message and move to the next Kafka message.

If  the `SKIP PARSER ERRORS` clause is not added during create pipeline, the default behavior will remain the same.

* Error messages will be logged in the `pipelines_errors` table, adding the Kafka message in the `LOAD_DATA_LINE` column.
* The pipeline will stop or skip the current batch and move to the next batch based on whether the `pipelines_stop_on_error` variable is set to true or false.
* The engine variable `pipelines_parse_errors_threshold` will stop or skip the current batch if the number of errors exceeds the threshold value. This depends on whether the `pipelines_stop_on_error` variable is set to true or false.
  > **📝 Note**: See [PIPELINES\_BATCHES](https://docs.singlestore.com/db/v9.1/reference/information-schema-reference/data-ingest/pipelines-batches-summary.md) for more details about batches.

***

Modified at: June 10, 2026

Source: [/db/v9.1/load-data/data-sources/load-data-from-kafka/kafka-pipeline-using-json-format/](https://docs.singlestore.com/db/v9.1/load-data/data-sources/load-data-from-kafka/kafka-pipeline-using-json-format/)

(An index of the documentation is available at /llms.txt)