CREATE PIPELINE

Syntax

CREATE [OR REPLACE] [AGGREGATOR] PIPELINE [IF NOT EXISTS] <pipeline_name> AS
  LOAD DATA { kafka_configuration | s3_configuration | filesystem_configuration
            | azure_blob_configuration | hdfs_configuration | gcs_configuration 
            | link_configuration | mongodb_configuration | mysql_configuration }
    [BATCH_INTERVAL <milliseconds>]
    [MAX_PARTITIONS_PER_BATCH <num_partitions>]
    [MAX_RETRIES_PER_BATCH_PARTITION <max_retries>]
    [MAX_OFFSETS_PER_BATCH_PARTITION <num_offsets>]
    [RESOURCE POOL <resource_pool_name>]
    [(ENABLE|DISABLE) OUT_OF_ORDER OPTIMIZATION]
    [WITH TRANSFORM ('uri', 'executable', 'arguments [...]') ]
  [REPLACE | IGNORE | SKIP { ALL | CONSTRAINT | DUPLICATE KEY | PARSER } ERRORS]
  [STOP_ON_ERROR { ON | OFF }]
  { INTO TABLE <table_name> | INTO PROCEDURE <procedure_name> }
  { <json_format_options> | <avro_format_options> | <parquet_format_options> | <csv_format_options> }
  [ (<column_name>, ... ) ]
  [SET  <column_name> = <expression>,... | <pipeline_source_file>]
  [WHERE <expression>,...]
  [ON DUPLICATE KEY UPDATE <column_name> = <expression>, [...]]
  [NULL DEFINED BY <string> [OPTIONALLY ENCLOSED]]

  <kafka_configuration>:
    KAFKA 'kafka_topic_endpoint'

  <s3_configuration>:
    S3 { '<bucket-name>' | '<bucket-name/path>' }
      [CONFIG '<configuration_json>']
      CREDENTIALS '<credentials_json>'

  <filesystem_configuration>:
    FS 'path'
      [CONFIG '<configuration_json>']

  <azure_blob_configuration>:
    AZURE { '<container-name>' | '<container-name/object-name>' | '<container-name/prefix/object-name>' }
      CREDENTIALS '<credentials_json>'
      [CONFIG '<configuration_json>']

  <hdfs_configuration>:
    HDFS '<hdfs://<namenode DNS> | IP address>:<port>/<directory>'
      [CONFIG '<configuration_json>']

  <gcs_configuration>:
   GCS { '<bucket-name>' | '<bucket-name/path>' }
     CREDENTIALS '<credentials_json>'
     [CONFIG '<configuration_json>']

  <link_configuration>:
   LINK <connection_name> '<path>'

  <mongodb_configuration>:
    MONGODB '<db_name.collection_name>' 
      CREDENTIALS '<credentials_json>'
      CONFIG '<configuration_json>'
      <avro_format_options> 

  <mysql_configuration>:
    MYSQL '<db_name.table_name>' 
      CREDENTIALS '<credentials_json>'
      CONFIG '<configuration_json>'
      <avro_format_options>

  <json_format_options>:
    FORMAT JSON
    ( {<column_name> | @<variable_name>} <- <subvalue_path> [DEFAULT <literal_expr>], ...)    
    [KAFKA KEY ( {<column_name> | @<variable_name>} <- <subvalue_path>     [DEFAULT <literal_expr>], ...)]
  
<avro_format_options>:
    FORMAT AVRO SCHEMA REGISTRY {uri}
    ( {<column_name> | @<variable_name>} <- <subvalue_path>, ...)
    [SCHEMA '<avro_schema>']
    [KAFKA KEY ( {<column_name> | @<variable_name>} <- <subvalue_path>, ...)]
    
  <subvalue_path>:
    {% | [%::]ident [::ident ...]}

  <parquet_format_options>:
    FORMAT PARQUET <parquet_subvalue_mapping> [TIMEZONE '<time_zone_name>']

  <parquet_subvalue_mapping>:
    ( {<column_name> | @<variable_name>} <- <subvalue_path>,, ...)

  <parquet_subvalue_path>:
    {ident [::ident ...]}

  <iceberg_format_options>:
    FORMAT ICEBERG <iceberg_subvalue_mapping>

  <iceberg_subvalue_mapping>:(
  {<singlestore_col_name> | @<variable_name>}<-<iceberg_subvalue_path> [, ... ])

  <iceberg_subvalue_path>: {ident [::ident ...]}

  <csv_format_options>:
    [FORMAT CSV]
    [{FIELDS | COLUMNS}
     TERMINATED BY '<string>'
       [[OPTIONALLY] ENCLOSED BY '<char>']
       [ESCAPED BY '<char>']
    ]
    [LINES
      [STARTING BY '<string>']
      [TERMINATED BY '<string>']
    ]
    [IGNORE <number> LINES]
    [ ({<column_name> | @<variable_name>}, ...) ]

Remarks

• SingleStore Helios does not support Filesystem Pipelines.

• Pipeline names are always case-sensitive for operations that refer to pipelines.

• If the OR REPLACE clause is provided and a pipeline with pipeline_name already exists, then the CREATE query will alter that pipeline to match the new definition (including its credentials). Its state, including its cursor positions, will be preserved by the CREATE.

• The OR REPLACE and IF NOT EXISTS clauses are mutually exclusive, i.e., they cannot be used in the same CREATE PIPELINE statement.

• The MONGODB and MYSQL source only support the AGGREGATOR pipelines and AVRO data type.

  Note: SingleStore recommends using the CREATE { TABLE | TABLES AS INFER PIPELINE } statements to generate pipeline templates for the MONGODB and MYSQL sources. Use the automatically generated templates to create the required pipelines. For related information, refer to Replicate MongoDB® Collections using SQL and Replicate Data from MySQL.

• CONFIG and CREDENTIALS can be specified in either order (CONFIG followed by CREDENTIALS or CREDENTIALS followed by CONFIG).

• SKIP CONSTRAINT ERRORS and SKIP DUPLICATE KEY ERRORS are unsupported with pipelines into stored procedures.

• IGNORE, SKIP PARSER ERRORS, and SKIP ALL ERRORS are supported with CSV and Kafka JSON pipelines only.

• REPLACE, SKIP CONSTRAINT ERRORS, and SKIP DUPLICATE KEY ERRORS are supported with non-CSV pipelines.

• This command causes implicit commits. Refer to COMMIT for more information.

• Pipelines can be used to ingest and process data with supported data sets. For more information, see Character Encoding.

• You can use the function pipeline_batch_id() in an expression in the SET clause. This function returns the id of the batch used to load the data. For example, given the table definition CREATE TABLE t(b_id INT, column_2 TEXT);, you could create this statement to load the batch id into the column b_id:

  SQL

  Copy

  CREATE PIPELINE p AS LOAD DATA ... INTO TABLE t(@b_id,column_2)
   ... SET b_id = pipeline_batch_id();

• CREATE PIPELINE with a 3rd party file storage system (S3, HDFS, Azure Blob, GSC, or Kafka) requires CREATE PIPELINE and OUTBOUND privileges.

  Without the required privileges, an error will be generated:

  SQL

  Copy

  CREATE PIPELINE:
  CREATE PIPELINE pipeline1 
      AS LOAD DATA S3 'my-bucket-name' 
      CONFIG '{"region": "us-east-1"}'
      CREDENTIALS '{"aws_access_key_id": "your_access_key_id", "aws_secret_access_key": "your_secret_access_key", "aws_session_token": "your_session_token"}'
      INTO TABLE 't1'
      FIELDS TERMINATED BY ',';
  ERROR 1227 (42000): Access denied; you need (at least one of) the OUTBOUND privilege(s) for this operation 

  Here is how to set the privileges for a user to be allowed to execute this command:

  SQL

  Copy

  CREATE PIPELINE:
  GRANT CREATE PIPELINE, OUTBOUND ON *.* to user1;

  The OUTBOUND privilege will be displayed in the SHOW GRANTS command:

  SQL

  Copy

  SHOW GRANTS FOR user1;

  +--------------------------------------------------------+
  | Grants for user1@%                                     |
  +--------------------------------------------------------+
  | GRANT CREATE PIPELINE, OUTBOUND ON *.* TO 'user1'@'%   |
  +--------------------------------------------------------+

• Pipeline names that begin with a number or are solely numeric; require backticks (` `) around the pipeline name.

  SQL

  Copy

  CREATE PIPELINE `3e93587cb1` AS LOAD DATA KAFKA 'host.example.com/my-topic' INTO TABLE <table_name>;

• STOP_ON_ERROR when enabled it will cause a pipeline to stop when an error occurs. If no value is specified, the behavior is controlled by the engine variable pipelines_stop_on_error. The syntax for enabling or disabling is as follows:

  SQL

  Copy

  CREATE PIPELINE <pipeline_name> AS LOAD DATA FS '</path/to/file' STOP_ON_ERROR ON/OFF INTO TABLE <table_name>;

• Refer to the Permission Matrix for the required permission.

REPLACE, IGNORE or SKIP ERRORS

REPLACE, IGNORE or SKIP can be used when creating a pipeline. See Additional CREATE PIPELINE Examples for more information.

The SKIP ... ERRORS behavior allows you to specify an error scenario that, when encountered, discards an offending row. See LOAD DATA for more information.

Unlike SKIP ... ERRORS which discards offending rows, IGNORE may change the inserted row’s data to ensure that it adheres to the table schema. See LOAD DATA

Creating a Pipeline Using a Connection Link

The CREATE PIPELINE .. LINK statement loads data from the data provider using a connection link. To use this command, you only need to know the connection link name, not the connection details and configuration. However, you need the SHOW LINK permission, provided by your administrator, to use a connection link. This command supports connections to S3, Azure, GCS, HDFS, and Kafka only.

The following example creates a pipeline using an HDFS connection:

USE db1;
CREATE PIPELINE mypipeline AS LOAD DATA
LINK HDFS connection 'hdfs://hadoop-namenode:8020/path/to/files'
INTO TABLE my_table;

START PIPELINE mypipeline;

Kafka Pipeline Syntax

For a tutorial on getting started with Kafka pipelines, see Load Data from Kafka Using a Pipeline.

The following example statement demonstrates how to create a Kafka pipeline using the minimum required syntax:

Minimum Required Kafka Pipeline Syntax

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

START PIPELINE pipeline_name;

This statement creates a new pipeline named pipeline_name, uses a Kafka workspace as the data source, points to the location of the my-topic topic at the Kafka workspace’s endpoint, and begins to ingest data into table_name when the pipeline is started.

CREATE PIPELINE pipeline_name AS LOAD DATA KAFKA '<Confluent Cloud workspace endpoint>/test'
CONFIG '{"sasl.username": "<CLUSTER_API_KEY>",
         "sasl.mechanism": "PLAIN",
         "security.protocol": "SASL_SSL",
       
       "ssl.ca.location":"/etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem",} 
CREDENTIALS '{"sasl.password": "<CLUSTER_API_SECRET>"}'
INTO TABLE table_name;

CREATE TABLE table_name (col_a INT, col_b TEXT);

CREATE PIPELINE pipeline_name AS
LOAD DATA KAFKA 'host.example.com/my-topic'
INTO TABLE 'table_name'
FORMAT JSON (col_a <- json_a)
KAFKA KEY (col_b <- json_b);

START PIPELINE pipeline_name;

message_headers = [	
(“Country”,”USA”),	
(“City”, “Arizona”),	
(“Language”, “English”),
]

CREATE PIPELINE <pipeline_name> AS
LOAD DATA KAFKA 'host1.example.com:9092,host2.example.com:9092,host3.example.com:9092/my-topic'
INTO TABLE 'table_name';

CREATE PIPELINE <pipeline_name> AS
LOAD DATA KAFKA 'host.example.com/my-topic'
MAX_OFFSETS_PER_BATCH_PARTITION = 30
INTO TABLE <table_name>;

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

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

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

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

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

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

S3 Pipeline Syntax

The following example statement demonstrates how to create an S3 pipeline using the minimum required syntax:

Minimum Required S3 Pipeline Syntax:

CREATE PIPELINE <pipeline_name> AS
LOAD DATA S3 'bucket-name'
CREDENTIALS '{"aws_access_key_id": "your_access_key_id", 
              "aws_secret_access_key": "your_secret_access_key"}'
INTO TABLE `table_name`;

START PIPELINE pipeline_name; 

This statement creates a new pipeline named pipeline_name, uses an S3 bucket named bucket-name as the data source, and will start ingesting the bucket’s objects into table_name. Credentials for your S3 bucket can either be in the form of an AWS access key or an Amazon Resource Name (ARN).

S3 pipelines support globbing (via the * wildcard) in the bucket path and name. For example, you could create this pipeline: CREATE PIPELINE p AS LOAD DATA S3 'my_bucket_path/file1*' ....

To load files from a specific folder in your S3 bucket while ignoring the files in the subfolders, use the '**' regular expression pattern as 's3://<bucket_name>/<folder_name>/**'. For example:

CREATE PIPELINE <your_pipeline> AS
LOAD DATA S3 's3://<bucket_name>/<folder_name>/**'
CONFIG '{"region":"<your_region>"}'
CREDENTIALS '{"aws_access_key_id": "<access_key_id>",  
              "aws_secret_access_key": "<secret_access_key>"}'
SKIP DUPLICATE KEY ERRORS
INTO TABLE <your_table>;

Using two asterisks (**) after the folder instructs the pipeline to load all of the files in the main folder and ignore the files in the subfolders. However, the files in the subfolders will get scanned when listing the contents of the bucket.

No CONFIG clause is required to create an S3 pipeline. However, this clause can be used to specify things, such as, the request_payer, endpoint_urlx, and more. For more information about SingleStore Helios-specific configurations for S3, refer to S3 Configurations.

As an alternative to specifying the CONFIG and CREDENTIALS clauses in a CREATE PIPELINE ... S3 statement, you can reference a connection link. For more information, see Configuring and Using Connection Links.

AWS Pipeline Syntax for AWS_ROLE_ARN and AWS_WEB_IDENTITY_TOKEN_FILE

AWS_ROLE_ARN and the contents of the AWS_WEB_IDENTITY_TOKEN_FILE may be used as credentials for creating a pipeline. A syntax example is below:

CREATE PIPELINE pipeline_web_token AS
LOAD DATA S3 'bucket-name'
CONFIG '{"region": "us-east-1"}'
CREDENTIALS '{"role_arn": "arn:aws:iam::...", 
              "aws_web_identity_token": "eyJhb...."}'
INTO TABLE `table_name`;

AWS Elastic Kubernetes Service  (EKS)  IAM Roles for Service Accounts (IRSA) Authentication

AWS EKS IRSA may be used when creating an S3 pipeline. To use AWS EKS IRSA credentials, contact SingleStore support. To enable this authentication add "creds_mode": "eks_irsa" to the CONFIG section of the pipeline and it will use the service role assigned to the EKS pods for S3 Authentication.

For a tutorial on getting started with S3 Pipelines, see Load Data from Amazon Web Services (AWS) S3.

S3 Pipeline Using Specified Region

CREATE PIPELINE pipeline_name AS
LOAD DATA S3 'bucket-name'
CONFIG '{"region": "us-west-1"}'
CREDENTIALS '{"aws_access_key_id": "your_access_key_id", 
              "aws_secret_access_key": "your_secret_access_key"
              [, "aws_session_token": "your_temp_session_token"][, "role_arn":"your_role_arn"]}'
INTO TABLE `table_name`;

S3 Pipeline Using Metadata Garbage Collection (GC)

By default, each pipeline maintains progress metadata for every known file name, for the lifetime of the pipeline. The optional ENABLE OFFSETS METADATA GC clause of CREATE PIPELINE causes the created pipeline to periodically discard metadata for already-loaded files, using a timestamp-based strategy that preserves exactly-once semantics. This option is only supported for S3 pipelines.

The number of batch metadata entries retained before being overwritten by new incoming batches may be adjusted by changing the default value of the global engine variable. pipelines_batches_metadata_to_keep. See Sync Variables Lists for more information.

CREATE PIPELINE <pipeline_name> AS LOAD DATA S3 "<file_name>" 
CONFIG '{"region": "us-west-1"}'
CREDENTIALS '{"aws_access_key_id": "your_access_key_id", 
              "aws_secret_access_key": "your_secret_access_key"
               [, "aws_session_token": "your_temp_session_token"][, "role_arn":"your_role_arn"]}'
ENABLE/DISABLE OFFSETS METADATA GC
INTO TABLE <table_name>;

Azure Blob Pipeline Syntax

The following example statement demonstrate how to create an Azure Blob pipeline using the minimum required syntax. Note that Azure Blob Pipelines are only available in MemSQL 5.8.5 and above.

Minimum Required Azure Pipeline Syntax:

CREATE PIPELINE pipeline_name AS
LOAD DATA AZURE 'container-name'
CREDENTIALS '{"account_name": "my_account_name", "account_key": "my_account_key"}'
INTO TABLE `table_name`;

START PIPELINE pipeline_name;

This statement creates a new pipeline named pipeline_name, uses an Azure Blob container named container-name as the data source, and will start ingesting the bucket’s objects into table_name. For a tutorial on getting started with Azure Pipelines, see Load Data from Azure Blob Storage Using a Pipeline.

Note that no CONFIG clause is required to create an Azure pipeline unless you need to specify the suffixes for files to load or the disable_gunzip condition. For more information about SingleStore Helios-specific configurations for Azure Blobs, refer to Azure Blob Configurations.

An Azure pipeline must authenticate with Azure before it can begin reading blobs from a container. The pipeline requires you to provide an account name and account access key.

The CREDENTIALS clause should be a JSON object with two fields:

account_name: this is the account name under which your blob container resides. This is usually a human-readable name, given by the person who created the account.

account_key: usually an 88 character 512-bit string linked to a storage account.

As an alternative to specifying the CONFIG and CREDENTIALS clauses in a CREATE PIPELINE ... AZURE statement, you can reference a connection link. For more information, see Configuring and Using Connection Links.

See the Azure documentation about viewing and managing Azure Access Keys to learn more.

HDFS Pipeline Syntax

The following example statement demonstrates how to create an HDFS pipeline using the minimum required syntax:

CREATE PIPELINE pipeline_name
AS LOAD DATA HDFS 'hdfs://hadoop-namenode:8020/path/to/files'
INTO TABLE `table_name`
FIELDS TERMINATED BY '\t';

START PIPELINE pipeline_name;

This example creates a new pipeline named pipeline_name and references the HDFS path /path/to/files as the data source. Once the pipeline is started, the HDFS extractor recursively walks the directory tree in the HDFS path, looking for MapReduce output files. Any files with names matching the regular expression part-(m|r)-(.*) are imported into my_table. The import occurs only if there are additional files in the directory that don’t match the regular expression; this check confirms that MapReduce created a SUCCESS file.

For more information on SingleStore Helios-specific configurations for HDFS, refer to HDFS Configurations.

As an alternative to specifying the CONFIG clause in a CREATE PIPELINE ... HDFS statement, you can reference a connection link. For more information, see Configuring and Using Connection Links.

For a tutorial on getting started with HDFS Pipelines, Load Data from HDFS Using a Pipeline.

GCS Pipeline Syntax

The following example statement demonstrates how to create a GCS pipeline using the minimum required syntax:

CREATE PIPELINE pipeline_name
AS LOAD DATA GCS 'bucket-name'
CREDENTIALS '{"access_id": "your_google_access_key", "secret_key": "your_google_secret_key"}'
INTO TABLE `table_name`
FIELDS TERMINATED BY ',';

START PIPELINE pipeline_name;

This example creates a new pipeline named pipeline_name, uses a GCS bucket named bucket-name as the data source, and ingests the bucket’s objects into my_table. A GCS pipeline requires authentication to Google before it can start reading objects from a bucket. It supports only HMAC keys and requires a GCS access key and secret key.

For more information on SingleStore Helios-specific configurations for GCS, refer to GCS Configurations.

GCS pipelines support globbing (via the * wildcard) in the bucket path and name. For example, you could create this pipeline: CREATE PIPELINE p AS LOAD DATA GCS 'my_bucket_path/file1*' ....

For a tutorial on getting started with GCS Pipelines, see Load Data from Google Cloud Storage (GCS) Using a Pipeline.

...
CREDENTIALS '{"access_id": "your_google_access_key", "secret_key": "your_google_secret_key"}'
...

Creating an Iceberg Pipeline

Refer to Iceberg Ingest for information on creating pipelines to directly ingest data from Apache Iceberg tables into SingleStore and for information on using <iceberg_format_options>.

Creating a Parquet Pipeline

The CREATE PIPELINE .. FORMAT PARQUET statement extracts specified fields from records in Parquet files. These files may be located in an Azure, HDFS, S3, or filesystem data source. Refer to Create a Parquet Pipeline for information on creating Parquet pipelines.

LOAD DATA

The SCHEMA REGISTRY {"IP" | "Hostname"} option allows LOAD DATA to pull an Avro schema from a schema registry. For more information, see the Avro Schema Evolution with Pipelines topic.

AS LOAD DATA: You can load data by specifying the data source as a Kafka workspace and topic, an S3 bucket or object (with optional prefix), or a file.

• AS LOAD DATA KAFKA 'topic_endpoint': To use Kafka as a data source, you must specify the endpoint for the Kafka workspace and the path to a specific topic with the workspace. For example:

  SQL

  Copy

  LOAD DATA KAFKA 'host.example.com/my-topic'

• AS LOAD DATA S3 'bucket-name': To use S3 as a data source, you must specify a bucket name or a bucket name and object name. For example:

  SQL

  Copy

  LOAD DATA S3 'bucket-name'

• AS LOAD DATA ... FORMAT AVRO SCHEMA REGISTRY {"IP" | "Hostname"}: This option allows the pipeline to pull an Avro schema from Confluent Schema Registry. You can optionally use the CONFIG and CREDENTIALS clauses to specify configuration settings to connect to the schema registry over SSL.

  Note

  The following syntax assumes you are creating a filesystem pipeline that connects to the schema registry over SSL. The ssl. settings shown only apply to SingleStore Helios 7.3.5 and later.

  SQL

  Copy

  CREATE PIPELINE ... AS LOAD DATA FS "/path/to/files/data.avro"
    INTO TABLE ...
    FORMAT AVRO
    SCHEMA REGISTRY "your_schema_registry_host_name_or_ip:your_schema_registry_port"
    ...
    CONFIG '{"schema.registry.ssl.certificate.location": "path-to-your-ssl-certificate-on-the-singlestore-db-node",
    "schema.registry.ssl.key.location": "path-to-your-ssl-certificate-key-on-the-singlestore-db-node",
    "schema.registry.ssl.ca.location": "path-to-your-ca-certificate-on-the-singlestore-db-node"}'
    CREDENTIALS '{"schema.registry.ssl.key.password": "your-password-for-the-ssl-certificate-key"}'

  Note

  You can use a subset of the ssl. settings as follows:

  • schema.registry.ssl.key.location, schema.registry.ssl.ca.location, and schema.registry.ssl.key.password

  • schema.registry.ssl.certificate.location, schema.registry.ssl.key.location, and schema.registry.ssl.key.password

  schema.registry.ssl.key.password is only required if your SSL certificate key has a password.

  As an alternative to specifying the SSL configuration settings in the CONFIG and CREDENTIALS clauses, you can install the registry's certificate (ca-cert) on all nodes in your SingleStore Heliosworkspace.

• [BATCH_INTERVAL milliseconds]: You can specify a batch interval in milliseconds, which is the amount of time that the pipeline waits before checking the data source for new data, once all of the existing data has been loaded from the data source. If a batch interval is not specified, the default value is 2500. For example:

  SQL

  Copy

  LOAD DATA KAFKA '127.0.0.1/my-topic'
  BATCH_INTERVAL 500

• [MAX_PARTITIONS_PER_BATCH max_partitions_per_batch]: Specifies the maximum number of batch partitions that can be scheduled into a single batch. Useful for limiting the parallelism of a pipeline on large workspaces to prevent the pipeline from throttling system resources.

• [MAX_RETRIES_PER_BATCH_PARTITION max_retries]: Specifies the maximum number of retries for each batch partition to write batch partition data to the destination table. If a batch transaction fails at any point, for example during extraction from a data source, optional transformation, or loading of the data into the destination table, it will be retried up to max_retries specified in this clause. If no value is specified or MAX_RETRIES_PER_BATCH_PARTITION is set to 0, then max_retries is set to the value specified by the PIPELINES_MAX_RETRIES_PER_BATCH_PARTITION engine variable.

• [RESOURCE POOL pool_name]: Specifies the resource pool that is used to load pipeline data.

  • If the resource pool is not specified at the time of pipeline creation, then the pipeline uses the user’s default resource pool. If no default resource pool has been set for the user, then the pipeline uses the value of the resource_pool engine variable as its resource pool.

    • In order to create a new resource pool, the user will need to have SUPER or CREATE_RG privileges.

    • Use information_schema.USER_PRIVILEGES to see users and their related privileges. See USER_PRIVILEGES for more information.

  • To obtain details about a resource pool, use select * from information_schema.mv_resource_pool_status to review the current running threads on a cluster for all nodes.

  • The background tasks of a pipeline runs in its resource pool. Therefore, a resource pool used by a pipeline cannot be dropped.

  • The user who creates the pipeline must have permissions to use the resource pool. The pipeline will continue to use the pool even if the right to use the pool for the pipeline is revoked for the user who created the pool.

  Note

  A good starting point for troubleshooting memory issues in pipelines is to run SELECT * FROM information_schema.mv_processlist. See mv_processlist for more information.

• [AGGREGATOR]: Specifying CREATE AGGREGATOR PIPELINE tells SingleStore Helios to pull data through the aggregator, instead of directly to the leaves. This option can be more efficient for low parallelism pipelines, like single file S3 loads or single partition Kafka topics (an aggregator pipeline is not required for single-partition Kafka topics). An aggregator pipeline is required for pipelines into reference tables and tables with auto increment columns. A stored procedure cannot be used with an aggregator pipeline.

echo "1,NULL,2020-08-06 07:53:09" >> /pipeline_test_infile/infile.csv
echo "2,2020-08-06 07:53:09,2020-09-06 07:53:09" > /pipeline_test_infile/infile.csv
echo "3,2020-08-06 07:53:09,NULL" >> /pipeline_test_infile/infile.csv

CREATE TABLE orders (
  ID INT,
  del_t1 DATETIME,
  del_t2 DATETIME);

CREATE PIPELINE order_load AS
    LOAD DATA FS '/pipeline_test_infile/'
    INTO TABLE orders
    FIELDS TERMINATED BY ','
    (ID, @del_t1, @del_t2)
    SET del_t1 = IF(@del_t1='NULL',NULL,@del_t1),
          del_t2 = IF(@del_t2='NULL',NULL,@del_t2);

TEST PIPELINE order_load;

+------+---------------------+---------------------+
| ID   | del_t1              | del_t2              |
+------+---------------------+---------------------+
|    1 | NULL                | 2020-08-06 07:53:09 |
|    2 | 2020-08-07 07:53:09 | 2020-09-08 07:53:09 |
|    3 | 2020-08-05 07:53:09 | NULL                |
+------+---------------------+---------------------+

CREATE TABLE keypairs(`key` VARCHAR(10), `value` VARCHAR(10)); 
CREATE PIPELINE jtest AS
LOAD DATA FS '<file path>/keypairs.json'
INTO TABLE keypairs
FORMAT JSON(`key` <- keypairs::`key`,`value` <- keypairs::`value`); 

START PIPELINE jtest;

SELECT * FROM keypairs;

+------+--------+
| key  | value  | 
+------+--------+
|    1 | 1      |
+------+--------+

CREATE TABLE teams(basketball VARCHAR(50), baseball VARCHAR (50), 
football VARCHAR(50), hockey VARCHAR(50));

CREATE PIPELINE teams_list AS
LOAD DATA FS '<file path>/jtinsert.json'
INTO TABLE teams
FORMAT JSON
(basketball <- teams::basketball,
baseball <- teams::baseball,
football <- teams::football,
hockey <- teams::hockey);

START PIPELINE teams_list;

SELECT * FROM teams;

+------------+-----------+----------+--------+
| basketball | baseball  | football | hockey |
+------------+-----------+----------+--------+
|  Lakers    | Dodgers   | Raiders  | Kings  |
+------------+-----------+----------+--------+

CREATE TABLE keypairs(`key` VARCHAR(10), `value` VARCHAR (10));

CREATE PIPELINE jtest AS
LOAD DATA S3 's3://<bucket_name>/<filename>.json'
CONFIG '{"region":"us-west-2"}'
CREDENTIALS '{"aws_access_key_id": "XXXXXXXXXXXXXXXXX",               
              "aws_secret_access_key": "XXXXXXXXXXX"}'
INTO TABLE keypairs
FORMAT JSON
(`key` <- keypairs::`key`,
`value` <- keypairs::`value`);

START PIPELINE jtest;
SELECT * FROM keypairs;

+------+--------+
| key  | value  | 
+------+--------+
|    1 | 1      |
+------+--------+

CREATE TABLE teams(basketball varchar(50), baseball varchar (50), 
football varchar(50), hockey varchar(50));

{
	"teams": [{
			"basketball": "Knicks"
		},
		{
			"baseball": "Yankees"
		},
		{
			"football": "Giants"
		},
		{
			"hockey": "Rangers"
		}
	]
}

CREATE PIPELINE teams_list AS 
LOAD DATA S3 's3://<bucket_name>/<filename>.json'
CONFIG '{"region":"us-west-2"}'
CREDENTIALS '{"aws_access_key_id": "XXXXXXXXXXXXXXXXX",               
              "aws_secret_access_key": "XXXXXXXXXXX"}'
INTO TABLE teams
FORMAT JSON
(basketball <- teams::basketball,
baseball <- teams::baseball,
football <- teams::football,
hockey <- teams::hockey);

START PIPELINE teams_list;
SELECT * FROM teams;

+------------+-----------+----------+--------+
| basketball | baseball  | football | hockey |
+------------+-----------+----------+--------+
|  Lakers    | Dodgers   | Raiders  | Kings  |
+------------+-----------+----------+--------+

CREATE TABLE employees(emp_id INT, data JSON);

csv file contents

emp_id,data
159,"{""name"": ""Damien Karras"", ""age"": 38, ""city"": ""New York""}"

CREATE OR REPLACE PIPELINE emps AS
    LOAD DATA FS '/tmp/<file_name>.csv'
    INTO TABLE employees
    FIELDS TERMINATED BY ','
    ENCLOSED BY '"'
    IGNORE 1 LINES;

START PIPELINE emps;

SELECT * FROM employees;

+--------+-----------------------------------------------------+
| emp_id | data                                                |
+--------+-----------------------------------------------------+
|    159 | {"age":38,"city":"New York","name":"Damien Karras"} |
+--------+-----------------------------------------------------+

csv file contents

emp_id,data
298,"{\"name\": \"Bill Denbrough\", \"age\": 25, \"city\": \"Bangor\"}"

CREATE OR REPlACE PIPELINE emps2 AS
    LOAD DATA FS '/tmp/<file_name>.csv'
    INTO TABLE employees
    FIELDS TERMINATED BY ',' 
    ENCLOSED BY '"' 
    ESCAPED BY '\\'
    IGNORE 1 LINES;

START PIPELINE emps2;

SELECT * FROM employees;

+--------+-----------------------------------------------------+
| emp_id | data                                                |
+--------+-----------------------------------------------------+
|    298 | {"age":25,"city":"Bangor","name":"Bill Denbrough"}  |
|    159 | {"age":38,"city":"New York","name":"Damien Karras"} |
+--------+-----------------------------------------------------+

csv file contents

emp_id,data
410,"{"name": "Annie Wilkes", "age": 45, "city":"Silver Creek"}"

CREATE OR REPLACE PIPELINE emps3 AS
    LOAD DATA FS '/tmp/<file_name>.csv'
    INTO TABLE employees
    FIELDS TERMINATED BY ',' 
    ENCLOSED BY '"'
    IGNORE 1 LINES;

START PIPELINE emps3;

ERROR 1262 (01000): Leaf Error (127.0.0.1:3307): Row 1 was truncated; it contained more data than there were input columns

csv file contents

emp_id,data
089,{"name": "Wilbur Whateley","age": 62,"city": "Dunwich"}

CREATE OR REPLACE PIPELINE emps4 AS
    LOAD DATA FS '/tmp/<file_name>.csv'
    INTO TABLE employees
    FIELDS TERMINATED BY ','
    ENCLOSED BY '{'
    IGNORE 1 LINES;

START PIPELINE emps4;

SELECT * FROM employees;

+--------+------------------------------------------------------+
| emp_id | data                                                 |
+--------+------------------------------------------------------+
|    298 | {"age":25,"city":"Bangor","name":"Bill Denbrough"}   |
|    159 | {"age":38,"city":"New York","name":"Damien Karras"}  |
|     89 | {"age":62,"city":"Dunwich","name":"Wilbur Whateley"} |
+--------+------------------------------------------------------+

INTO PROCEDURE

Creates a pipeline that uses a stored procedure to shape the data that is extracted from the pipeline's data source. For more information, see CREATE PIPELINE ... INTO PROCEDURE.