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

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.

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

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

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 for more information.

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 for more details about batches.