INFER PIPELINE

Syntax

INFER PIPELINE AS LOAD DATA {input_configuration}
     [FORMAT [CSV | JSON | AVRO | PARQUET | ICEBERG]]
     [AS JSON]

Remarks

• The input_configuration specifies configuration for loading files from Apache Kafka, Amazon S3, a local filesystem, Microsoft Azure, HDFS, and Google Cloud Storage. Refer to CREATE PIPELINE for more information on configuration specifications.

• All options supported by CREATE PIPELINE are supported by INFER PIPELINE.

• CSV, JSON, Avro, Parquet , and Iceberg formats are supported.

• The default format is CSV.

• TEXT and ENUM types use utf8mb4 charset and utf8mb4_bin collation by default.

• The AS JSON keyword is used to produce pipeline and table definitions in JSON format.

• For LOAD DATA CONNECTOR sources, INFER PIPELINE generates a CREATE TABLE and matching CREATE PIPELINE statement based on the connector's schema.

• When the connector is recognized, INFER PIPELINE appends connector-specific fields to the generated CONFIG. This is currently implemented only for KinesisSourceConnector, which adds:

  • "kafka.topic": "s2-kinesis": Kinesis emits into a single topic; this field is not user-relevant but the connector requires it.

  • "output.topic": "false": Disables the topic field in the output document.

• INFER PIPELINE for FORMAT JSON operates only on first-level fields of the connector's output. To ingest sub-fields (for example, value::data), edit the generated DDL manually or use an output mode to project the payload.

• INFER PIPELINE for FORMAT AVRO derives column names from Avro record leaf nodes. If two leaves share a name, the generated column name uses the full path joined by "." (for example, key.partitionKey).

• CREATE INFERRED PIPELINE does not support IF NOT EXISTS on the generated table. If you drop the pipeline and re-run INFER PIPELINE for a pipeline with the same name, the second CREATE TABLE fails with ER_TABLE_EXISTS_ERROR. Drop the table manually, or use CREATE PIPELINE (not INFERRED) when the table already exists.

Example

The following example demonstrates how to use the INFER PIPELINE command to infer the schema of a Avro-formatted file in an AWS S3 bucket.

This example uses data that conforms to the schema of the books table, as shown in the following.

{"namespace": "books.avro",
"type": "record",
"name": "Book",
"fields": [
{"name": "id", "type": "int"},
{"name": "name", "type": "string"},
{"name": "num_pages", "type": "int"},
{"name": "rating", "type": "double"},
{"name": "publish_timestamp", "type": "long",
"logicalType": "timestamp-micros"} ]}

Refer to Generate an Avro File for an example of generating an Avro file that conforms to this schema.

The following example generates a table and pipeline definition by scanning the specified Avro file and inferring the schema from selected rows. The output is displayed in query definition format.

INFER PIPELINE AS LOAD DATA S3
        's3://data_folder/books.avro'
CONFIG '{"region":"<region_name>"}'
CREDENTIALS '{
    "aws_access_key_id":"<your_access_key_id>",
    "aws_secret_access_key":"<your_secret_access_key>",
    "aws_session_token":"<your_session_token>"}'
FORMAT AVRO;

"CREATE TABLE `infer_example_table` (
    `id` int(11) NOT NULL,
    `name` longtext CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,
    `num_pages` int(11) NOT NULL,
    `rating` double NULL,
    `publish_date` bigint(20) NOT NULL);
CREATE PIPELINE `infer_example_pipeline`
AS LOAD DATA S3 's3://data-folder/books.avro'
CONFIG '{\""region\"":\""us-west-2\""}'
CREDENTIALS '{\n    \""aws_access_key_id\"":\""your_access_key_id\"",
\n    \""aws_secret_access_key\"":\""your_secret_access_key\"",
\n    \""aws_session_token\"":\""your_session_token\""}'
BATCH_INTERVAL 2500
DISABLE OUT_OF_ORDER OPTIMIZATION
DISABLE OFFSETS METADATA GC
INTO TABLE `infer_example_table`
FORMAT AVRO(
    `id` <- `id`,
    `name` <- `name`,
    `num_pages` <- `num_pages`,
    `rating` <- `rating`,
    `publish_date` <- `publish_date`);"

Refer to Schema and Pipeline Inference - Examples for more examples.

The following example uses INFER PIPELINE to generate a table and pipeline for an Amazon Kinesis stream. The generated CONFIG includes the connector-specific fields described in Remarks.

CREATE INFERRED PIPELINE kinesis_orders
AS LOAD DATA CONNECTOR 'KinesisSourceConnector'
CONFIG '{
  "kinesis.stream": "orders-stream",
  "kinesis.region": "us-east-1",
  "tasks.max": "3"
}'
CREDENTIALS '{
  "aws.access.key.id": "<ACCESS_KEY>",
  "aws.secret.access.key": "<SECRET_KEY>"
}';

The command produces a CREATE TABLE matching the connector's output schema and a CREATE PIPELINE statement whose CONFIG includes the appended fields:

{
  "kafka.topic": "s2-kinesis",
  "kinesis.stream": "orders-stream",
  "kinesis.region": "us-east-1",
  "tasks.max": "3",
  "output.topic": "false"
}

For more information on FORMAT, output modes, and column mapping, refer to Connector Pipelines.