Data Pipelines

CREATE [ OR REPLACE ] [ BATCH | CONTINUOUS | TRANSACTIONAL ] PIPELINE [ IF NOT EXISTS ] pipeline_name
    [ <advanced_pipeline_options> ]
    [ BAD_DATA_TARGET <kafka_bad_data_target> ]

    SOURCE ( <s3_source> | <filesystem_source> | <hdfs_source> | <kafka_source> )
        [ MONITOR ( <sqs_monitor> | <kafka_monitor> ) ]
    [ LOOKUP lookup_source <lookup_options> ]  [ ... ]

    EXTRACT
        ( <delimited_extract_options> | <asn1_extract_options> |
        <avro_extract_options> | <json_extract_options> |
        <parquet_extract_options> | <binary_extract_options> |
        <xml_extract_options> )
        [ <general_extract_options> ]

    [ INSERT ] INTO destination_table_name

    SELECT
        expression as col_alias_target,
        expression2 as col_alias_target2,
        ...
    [ WHERE filter_expression ]

    [ [ INSERT ] INTO destination_table_name_n
    SELECT
        expression_n as col_alias_target_n,
        expression2_n as col_alias_target2_n,
        ...
    [ WHERE filter_expression_n ] ] [ ... ]

/********************/
/* ADVANCED OPTIONS */
/********************/
advanced_pipeline_options ::=
    [ CORES processing_cores ]
    [ PARTITIONS file_partitions ]
    [ BATCH_SIZE number_of_rows ]
    [ RECORD_NUMBER_FORMAT record_number_format ]

/***************************/
/* BAD_DATA_TARGET OPTIONS */
/***************************/
kafka_bad_data_target ::=
    KAFKA
        BOOTSTRAP_SERVERS bootstrap_servers
        TOPIC topic_name
        [ CONFIG config_option ]

/******************/
/* SOURCE OPTIONS */
/******************/
s3_source ::=
    S3
        <file_based_source_options>
        BUCKET bucket_name
        ( FILTER | FILTER_GLOB | FILTER_REGEX | OBJECT_KEY ) specifiers
        [ REGION region ]
        [ ENDPOINT endpoint ]
        [ ENABLE_PATH_STYLE_ACCESS enable_path_style_access ]
        [ ACCESS_KEY_ID access_key_credentials ]
        [ SECRET_ACCESS_KEY secret_key_credentials ]
        [ SESSION_TOKEN session_token ]
        [ ROLE_ARN role_arn ]
        [ ASSUME_ROLE_CONFIG assume_role_config]
        [ MAX_CONCURRENCY parallel_connections ]
        [ READ_TIMEOUT num_seconds ]
        [ REQUEST_DEPTH num_requests ]
        [ REQUEST_RETRIES num_retries ]
        [ HEADERS headers ]

filesystem_source ::=
    FILESYSTEM
        <file_based_source_options>
        ( FILTER | FILTER_GLOB | FILTER_REGEX ) specifiers

hdfs_source ::=
    HDFS
        <file_based_source_options>
        ( FILTER | FILTER_GLOB | FILTER_REGEX ) specifiers
        ENDPOINT endpoint
        [ CONFIG hdfs_config ]

kafka_source ::=
    KAFKA
        BOOTSTRAP_SERVERS bootstrap_servers
        TOPIC topic_name
        [ WRITE_OFFSETS write_offsets ]
        [ CONFIG config_option ]
        [ AUTO_OFFSET_RESET ( 'latest' | 'earliest' )

file_based_source_options ::=
    [ PREFIX prefix ]
    [ COMPRESSION_METHOD 'gzip' ]
    [ SORT_BY ( 'filename' | 'created' | 'modified' )
        [ SORT_DIRECTION ( ASC | DESC ) ] ]
    [ SORT_REWRITE sort_rewrite ]
    [ START_FILENAME start_filename ]
    [ END_FILENAME end_filename ]
    [ START_CREATED_TIMESTAMP start_created_timestamp ]
    [ END_CREATED_TIMESTAMP end_created_timestamp ]
    [ START_MODIFIED_TIMESTAMP start_modified_timestamp ]
    [ END_MODIFIED_TIMESTAMP end_modified_timestamp ]

/******************/
/* MONITOR OPTIONS */
/******************/
sqs_monitor ::=
    SQS
        <general_monitor_options>
        QUEUE_URL queue_url
        [ REGION region ]
        [ ENDPOINT endpoint ]
        [ ACCESS_KEY_ID access_key_id ]
        [ SECRET_ACCESS_KEY secret_access_key ]

kafka_monitor ::=
    KAFKA
        <general_monitor_options>
        BOOTSTRAP_SERVERS bootstrap_servers
        TOPIC topic
        AUTO_OFFSET_RESET ( 'latest' | 'earliest' )
        [ GROUP_ID group_id ]
        [ MESSAGE {
            FILENAME file_selector
            TIMESTAMP timestamp_selector
            SIZE size_selector
        } ]
        [ CONSUMER {
            MAX_MESSAGES max_messages
            TIMEOUT consumer_timeout
        } ]

general_monitor_options ::=
    [ POLLING_INTERVAL polling_interval ]
    [ BATCH {
        MAX_FILES max_files
        TIMEOUT batch_timeout
        LOOKBACK lookback
     } ]

/*******************/
/* LOOKUP OPTIONS  */
/*******************/
lookup_options ::=
    CONNECTION_TYPE connection_type
    CONNECTION_STRING connection_string
    LOOKUP_SCHEMA lookup_schema
    LOOKUP_TABLE lookup_table
    [ REFRESH_PERIOD refresh_period ]
    [ CONFIG config_json ]

/*******************/
/* EXTRACT OPTIONS */
/*******************/
general_extract_options ::=
    [ CHARSET_NAME charset_name ]
    [ COLUMN_DEFAULT_IF_NULL column_default_if_null ]
    [ NULL_STRINGS null_strings ]
    [ TRIM_WHITESPACE trim_whitespace ]
    [ VALIDATE_CHARACTERS validate_characters ]
    [ REPLACE_INVALID_CHARACTERS replace_invalid_characters ]
    [ REPLACEMENT_CHARACTER replacement_character ]

delimited_extract_options ::=
    FORMAT ( 'delimited' | 'csv' )
        [ COMMENT_CHAR comment_char ]
        [ EMPTY_FIELD_AS_NULL empty_field_as_null ]
        [ ESCAPE_CHAR escape_char ]
        [ FIELD_DELIMITER field_delimiter ]
        [ FIELD_OPTIONALLY_ENCLOSED_BY enclosure_char ]
        [ HEADERS delimited_headers ]
        [ NUM_HEADER_LINES num_header_lines ]
        [ NUM_FOOTER_LINES num_footer_lines ]
        [ RECORD_DELIMITER record_delimiter ]
        [ SKIP_EMPTY_LINES skip_empty_lines ]
        [ OPEN_ARRAY open_array_char ]
        [ CLOSE_ARRAY close_array_char ]
        [ ARRAY_ELEMENT_DELIMITER array_delimiter ]
        [ OPEN_OBJECT open_object_char ]
        [ CLOSE_OBJECT close_object_char ]
        [ STRIP_ARRAY_ELEMENT_QUOTES strip_array_element_quotes ]
        [ STRIP_FIELD_QUOTES strip_field_quotes ]
        [ TRIM_ARRAY_ELEMENTS trim_array_elements]

asn1_extract_options ::=
    FORMAT 'asn.1'
        SCHEMA {
            URL url_asn1
            RECORD_TYPE record_type
        }

avro_extract_options ::=
    FORMAT 'avro'
        [ SCHEMA {
            [ INLINE inline_string ]
            [ INFER_FROM 'sample_file' ]
        } ]

binary_extract_options ::=
    FORMAT 'binary'
        RECORD_LENGTH length_in_bytes
        [ ENDIANNESS ( 'big' | 'little' ) ]
        [ AUTO_TRIM_PADDING auto_trim_padding ]
        [ PADDING_CHARACTER padding_character ]

json_extract_options ::=
    FORMAT 'json'

parquet_extract_options ::=
    FORMAT 'parquet'
        SCHEMA {
            INFER_FROM infer_from
        }

xml_extract_options ::=
    FORMAT 'xml'

...
FORMAT json
INTO public.orders
SELECT
    TIMESTAMP(BIGINT($created_timestamp)) as ordertime,
    metadata('filename') as source_filename,
    $order_number as ordernumber,
    $customer.first_name as fname,
    LEFT($customer.middle_initial,1) as minitial,
    $customer.last_name as lname,
    $postal_code as postal_code,
    $promo_code as promo_code,
    $order_total as ordertotal,
    DECIMAL($tax,8,2) as tax,
    CHAR[]($line_items[].product_name) as product_names,
    CHAR[]($line_items[].sku) as skus

...
FORMAT json
INTO public.orders
SELECT
    TIMESTAMP(BIGINT($created_timestamp)) as ordertime,
    metadata('filename') as source_filename,
    $order_number as ordernumber,
    $customer.first_name as fname,
    LEFT($customer.middle_initial,1) as minitial,
    $customer.last_name as lname,
    $postal_code as postal_code,
    $promo_code as promo_code,
    $order_total as ordertotal,
    DECIMAL($tax,8,2) as tax,
    CHAR[]($line_items[].product_name) as product_names,
    CHAR[]($line_items[].sku) as skus
WHERE
    STARTSWITH(COALESCE($customer.last_name, ''),'a')

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;

CREATE PIPELINE orders_pipeline
    SOURCE S3
        ENDPOINT 'https://s3.us-east-1.amazonaws.com'
        BUCKET 'ocient-docs'
        FILTER 'metabase_samples/csv/orders.csv'
        ENABLE_PATH_STYLE_ACCESS true
    EXTRACT
        FORMAT csv
        NUM_HEADER_LINES 1
INTO public.orders
SELECT
    $1 as id,
    $2 as user_id,
    $3 as product_id,
    $4 as subtotal,
    $5 as tax,
    $6 as total,
    $7 as discount,
    $8 as created_at,
    $9 as quantity;

CREATE CONTINUOUS PIPELINE test_continuous_pipeline
    PARTITIONS 32
    CORES 16
    SOURCE S3
        ENDPOINT 'http://endpoint.ocient.com/'
        BUCKET 'cs_data'
        FILTER '*.csv'
        MONITOR kafka
            BOOTSTRAP_SERVERS 'test-broker:9092'
            TOPIC 'cfl_kafka_ten_adtech_flat_small'
            AUTO_OFFSET_RESET 'earliest'
            GROUP_ID '84079bf1-bdc4-4b10-ba12-41ba6b17dffe'
    EXTRACT
        FORMAT csv
        RECORD_DELIMITER '\n'
    INTO public.ad_sessions
SELECT
    TO_TIMESTAMP(CHAR($1), 'yyyy-MM-dd HH:mm:ss.SSSSSS', 'java') AS event_date_time,
    CHAR($2)                    AS device_model,
    TINYINT($8)                 AS device_user_age,
    BOOLEAN($10)                AS device_ad_tracking_disabled,
    BINARY($11)                 AS device_mac,
    INT($14)                    AS ip_zip,
    FLOAT($19)                  AS ip_zip_latitude,
    DOUBLE($20)                 AS ip_zip_longitude,
    BIGINT($21)                 AS session_id,
    SMALLINT($32)               AS session_response_latency,
    DECIMAL($34, 10, 1)         AS session_transaction_revenue,
    CHAR($39)                   AS session_app_name
;

CREATE TRANSACTIONAL PIPELINE test_continuous_pipeline
    SOURCE S3
        ENDPOINT 'http://endpoint.ocient.com/'
        BUCKET 'cs_data'
        FILTER '*.csv'
    EXTRACT
        FORMAT csv
    INTO public.ad_sessions
SELECT
    TO_TIMESTAMP(CHAR($1), 'yyyy-MM-dd HH:mm:ss.SSSSSS', 'java') AS event_date_time,
    CHAR($2)                    AS device_model,
    TINYINT($8)                 AS device_user_age,
    BOOLEAN($10)                AS device_ad_tracking_disabled,
    BINARY($11)                 AS device_mac,
    INT($14)                    AS ip_zip,
    FLOAT($19)                  AS ip_zip_latitude,
    DOUBLE($20)                 AS ip_zip_longitude,
    BIGINT($21)                 AS session_id,
    SMALLINT($32)               AS session_response_latency,
    DECIMAL($34, 10, 1)         AS session_transaction_revenue,
    CHAR($39)                   AS session_app_name
;

CREATE PIPELINE ...
    SOURCE FILESYSTEM
        PREFIX '/tmp/sample-data/'
        FILTER '**/*.csv'
    EXTRACT
        FORMAT delimited
INTO public.orders
SELECT
    $1 as username,
    $2 as subtotal,
    ...

CREATE PIPELINE ...
    BAD_DATA_TARGET
        KAFKA
        TOPIC 'orders_errors'
        BOOTSTRAP_SERVERS '111.11.111.1:9092,111.11.111.2:9092'
        CONFIG '{"compression.type": "gzip"}'
    SOURCE
        ...
    EXTRACT
        ...
INTO public.orders
SELECT
    $order.billing.name as username,
    $order.subtotal as subtotal,
    ...

DROP PIPELINE [ IF EXISTS ] pipeline_name [, ...]

DROP PIPELINE ad_data_pipeline;

DROP PIPELINE IF EXISTS ad_data_pipeline;

PREVIEW PIPELINE pipeline_name
    [ MODE mode ]
    [ SHOW_ERRORS_AS_JSON show_errors_as_json ]
    SOURCE [ INLINE ] (inline_string | <s3_source> | <filesystem_source> | <kafka_source>)
        [ LIMIT limit ]
    EXTRACT
        FORMAT csv
        RECORD_DELIMITER record_delimiter
        FIELD_DELIMITERS ['delim1', 'delim2', ...]
        [ INTERMEDIATE_VALUES intermediate_values ]
[ INSERT ] INTO created_tablename
SELECT
    preview_column_formula AS preview_column_name, ...
[ WHERE filter_expression ]
[ [ INSERT ] INTO created_tablename_n
SELECT
    preview_column_formula AS preview_column_name, ...
[ WHERE filter_expression ] ] [ ,... ]
[ FOR created_tablename_n ]

CREATE TABLE previewload (col1 VARCHAR, col2 INT, col3 BOOLEAN);

PREVIEW PIPELINE testpipeline
    SOURCE INLINE 'hello,2,true|bye,3,false'
    EXTRACT
        FORMAT CSV
        RECORD_DELIMITER '|'
        FIELD_DELIMITERS [',']
INTO previewload
SELECT
    $1 AS col1,
    $2 AS col2,
    $3 AS col3;

col1                                         col2       col3
\--------------------------------------------------------------
hello                                        2          true
bye                                          3          false

Fetched 2 rows

DROP TABLE previewload;

CREATE TABLE previewload (col1 VARCHAR, col2 INT, col3 BOOLEAN);

PREVIEW PIPELINE testpipeline
    SOURCE INLINE e'hello\tworld,2,true|bye\tworld,3,false'
    EXTRACT
        FORMAT CSV
        RECORD_DELIMITER '|'
        FIELD_DELIMITERS [',']
    INTO previewload
    SELECT
        $1 AS col1,
        $2 AS col2,
        $3 AS col3;

col1                                         col2       col3
\--------------------------------------------------------------
hello   world                                  2        true
bye     world                                  3        false

Fetched 2 rows

DROP TABLE previewload;

CREATE TABLE previewload (col1 VARCHAR, col2 VARCHAR, col3 VARCHAR);

PREVIEW PIPELINE testpipeline
    SOURCE INLINE 'hello,world|bye,world'
    EXTRACT
        FORMAT CSV
        RECORD_DELIMITER '|'
        FIELD_DELIMITERS [',']
    INTO previewload
    SELECT
        $1 AS col1,
        $2 AS col2,
        CONCAT($1,$2) AS col3;

col1                                         col2                                         col3
\---------------------------------------------------------------------------------------------------------------------------------------
hello                                        world                                        helloworld
bye                                          world                                        byeworld

Fetched 2 rows

DROP TABLE previewload;

CREATE TABLE previewload (
    id INT NOT NULL,
    salut VARCHAR(3) NOT NULL,
    name VARCHAR(10) NOT NULL,
    surname VARCHAR(10) NOT NULL,
    zipcode INT NOT NULL,
    age INT NOT NULL,
    rank INT NOT NULL);

PREVIEW PIPELINE test_small_kafka_simple_csv
    SOURCE KAFKA
        TOPIC 'ddl_csv'
        WRITE_OFFSETS false
        BOOTSTRAP_SERVERS 'servername:0000'
        CONFIG '{"auto.offset.reset": "earliest"}'
        LIMIT 3
    EXTRACT
        FORMAT csv
        RECORD_DELIMITER '\n'
    INTO previewload
    SELECT
        INT($1) AS id,
        CHAR($2) AS salut,
        CHAR($3) AS name,
        CHAR($4) AS surname,
        INT($5) AS zipcode,
        INT($6) AS age,
        INT($7) AS rank;

id         salut                                        name                                         surname                                      zipcode    age        rank
\-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
105        Mr                                           Jmhsuxofspx                                  Uaofgayjugb                                  85573      29         2
101        Mr                                           Ijmmtbddkyh                                  Yqbxqnkgidp                                  52393      43         1
109        Mr                                           Bigohpwfwmr                                  Qcxgakpkoeu                                  74420      1          3

Fetched 3 rows

START PIPELINE pipeline_name
    [ ERROR
        [ LIMIT <integer_value> ]
        [ FILE_ERROR (FAIL | SKIP_MISSING_FILE | TOLERATE) ] ]
    [ USING LOADERS <loader_names> ]
    [ ON COMPLETION (NO_FLUSH | FLUSH_AND_WAIT | FLUSH_AND_RETURN) ]

START PIPELINE ad_data_pipeline;

START PIPELINE ad_data_pipeline
    ERROR LIMIT 10;

START PIPELINE ad_data_pipeline
    USING LOADERS "stream-loader1";

STOP PIPELINE pipeline_name

STOP PIPELINE ad_data_pipeline;

ALTER PIPELINE [ IF EXISTS ] pipeline_name
[ INSERT ] INTO table_name
SELECT
    $field_name AS field_name [ , ... ]
[ WHERE ... ]
[ INSERT INTO table_nameN SELECT ... ]
[ FORCE ]

CREATE TABLE users(
    id         UUID NOT NULL,
    firstname  VARCHAR(255) NOT NULL,
    lastname   VARCHAR(255) NOT NULL,
    birthyear  INT,
    groups     VARCHAR(255)[] NOT NULL DEFAULT 'char[]'
);

CREATE PIPELINE users_pipeline
    SOURCE filesystem
        FILTER '/data/users/*.avro'
    EXTRACT
        FORMAT avro
        SCHEMA {
            INFER_FROM all_files
        }
INTO users
SELECT
    $id         AS id,
    $firstname  AS firstname,
    $lastname   AS lastname,
    $birthyear  AS birthyear;

START PIPELINE users_pipeline;

STOP PIPELINE users_pipeline;

ALTER PIPELINE users_pipeline
INTO users
SELECT
    $id         AS id,
    $firstname  AS firstname,
    $lastname   AS lastname,
    $birthyear  AS birthyear,
    $groups[]   AS groups;

START PIPELINE users_pipeline;

STOP PIPELINE users_pipeline;

ALTER PIPELINE IF EXISTS users_pipeline
INTO users
SELECT
    $id         AS id,
    $firstname  AS firstname,
    $lastname   AS lastname,
    $birthyear  AS birthyear,
    $groups[]   AS groups;

START PIPELINE users_pipeline;

STOP PIPELINE users_pipeline;

ALTER PIPELINE users_pipeline
INTO users
SELECT
    $id         AS id,
    $firstname  AS firstname,
    $lastname   AS lastname;

START PIPELINE users_pipeline;

STOP PIPELINE users_pipeline;

ALTER PIPELINE users_pipeline
INTO users
SELECT
    $id         AS id,
    $firstname  AS firstname,
    $lastname   AS lastname,
    SMALLINT($birthyear)  AS birthyear;

START PIPELINE users_pipeline;

STOP PIPELINE users_pipeline;

ALTER PIPELINE users_pipeline
INTO users
SELECT
    $id         AS id,
    $firstname  AS firstname,
    $lastname   AS lastname,
    $birthyear  AS birthyear,
WHERE
    $birthyear > 1950
FORCE;

START PIPELINE users_pipeline;

ALTER PIPELINE [ IF EXISTS ] pipeline_original_name RENAME TO pipeline_new_name

ALTER PIPELINE ad_data_pipeline RENAME TO renamed_pipeline;

EXPORT PIPELINE pipeline_name

EXPORT PIPELINE ad_data_pipeline;

CREATE [ OR REPLACE ] PIPELINE FUNCTION [ IF NOT EXISTS ]
    function_name( input_argument [, ...] )
    LANGUAGE GROOVY
    RETURNS output_argument_definition
    IMPORTS [ library_name [, ...] ]
    AS $$
        groovy_declaration
    $$

CREATE PIPELINE FUNCTION
    sort_function( value INT[] NOT NULL, ascending BOOLEAN NOT NULL)
    LANGUAGE GROOVY
    RETURNS INT[] NOT NULL
    IMPORTS [
        'java.lang.Integer',
        'java.util.ArrayList',
        'java.util.Collections',
        'java.util.Comparator',
        'java.util.List'
        ]
    AS $$
        /* Throw an error if the array is empty */
        if (value.isEmpty()){
            throw new PipelineFunctionException("Unexpected empty array");
        }

        /* Make a copy of the list. */
        List<Integer> sorted = new ArrayList<>((List<Integer>)value);

        /* Sort the array elements according to the specified order. */
        sorted.sort(ascending ? Comparator.naturalOrder() : Comparator.reverseOrder());

        /* Return the sorted array. */
        return sorted;
    $$;

SELECT name, return_type, argument_names, argument_types, imported_libraries
FROM sys.pipeline_functions;

DROP PIPELINE FUNCTION [ IF EXISTS ] function_name [, ...]

DROP PIPELINE FUNCTION sort_function;

DROP PIPELINE FUNCTION IF EXISTS sort_function;