Data Formats for Data Pipelines

Example DEFINITIONS IMPLICIT TAGS ::=
BEGIN
    PersonnelRecord ::= [APPLICATION 0] SET {
		name         [0] Name,
		title        [1] VisibleString,
		number       [2] EmployeeNumber,
		dateOfHire   [3] Date,
		nameOfSpouse [4] Name,
		children     [5] SEQUENCE OF ChildInformation DEFAULT {}
    }
    ChildInformation ::= SET {
		name        [0] Name,
		dateOfBirth [1] Date
    }
    Name ::= [APPLICATION 1] SEQUENCE {
		givenName  [0] VisibleString,
		initial    [1] VisibleString,
		familyName [2] VisibleString
    }
    EmployeeNumber ::= [APPLICATION 2] INTEGER
    Date ::= [APPLICATION 3] VisibleString
END

CREATE TABLE personnel_records(
    first_name VARCHAR NOT NULL,
    initial VARCHAR NOT NULL,
    family_name VARCHAR NOT NULL,
    title VARCHAR NOT NULL,
    number BIGINT NOT NULL,
    date_of_hire DATE NOT NULL
);

CREATE PIPELINE personnel_pipeline
    SOURCE S3
        BUCKET 'misc'
        ENDPOINT 'http://cos'
        ACCESS_KEY_ID ''
        SECRET_ACCESS_KEY ''
        FILTER_GLOB 'user/asn1/personnel_records.der'
    EXTRACT
        FORMAT 'asn.1'
        SCHEMA {
            URL 'http://cos/filepath/asn1/personnel_records.asn'
            RECORD_TYPE 'Example.PersonnelRecord'
        }
INTO personnel_records
SELECT
    $name.givenName AS first_name,
    $name.initial as initial,
    $name.familyName as family_name,
    $title as title,
    $number as number,
    TO_DATE($dateOfHire, 'yyyyMMdd') as date_of_hire;

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 'sample_file'
    }
INTO users
SELECT
    $id         AS id,
    $firstname  AS firstname,
    $lastname   AS lastname,
    $birthyear  AS birthyear,
    $groups[]   AS groups;

CREATE PIPELINE users_pipeline
SOURCE filesystem
    FILTER '/data/users/*.avro'
EXTRACT
    FORMAT avro
    SCHEMA {
        INLINE '{
            "type": "record",
            "name": "User",
            "namespace": "test.users",
            "fields": [
                { "name": "id",        "type": { "type": "string", "logicalType": "uuid" } },
                { "name": "firstname", "type": "string" },
                { "name": "lastname",  "type": "string" },
                { "name": "birthyear", "type": ["null","int"], "default": null },
                { "name": "groups",    "type": { "type": "array", "items": "string" } }
            ]
        }'
    }
INTO users
SELECT
    $id         AS id,
    $firstname  AS firstname,
    $lastname   AS lastname,
    $birthyear  AS birthyear,
    $groups[]   AS groups;

CREATE PIPELINE binary_users_pipeline
SOURCE
	S3 ...
FORMAT
	BINARY
	RECORD_LENGTH 62
	CHARSET_NAME 'cp500'
INTO public.users
SELECT
	$"[1, 20]" as first_name,
	$"[21, 20]" as last_name,
	INT($"[41, 4]") as age,
	DECIMAL($"[45, 10]", 'packed', 2) as total_spent,
	BIGINT($"[55, 8]", 'unsigned', 'little') as user_id;

iphone|60607|viewed|502|293.99|[shopping,news]

CREATE PIPELINE delimited_pipeline
SOURCE s3
    ...
FORMAT delimited
    FIELD_DELIMITERS ['|']
    ...
SELECT
    CHAR($1) as device_model,
    INT($2) as zip,
    INT($3) as amount,
    DOUBLE($5) as price,
    CHAR[]($6[]) as categories;

CREATE TABLE test_table (
  col_a INT NULL,
  col_b VARCHAR NULL,
  col_c VARCHAR NULL
);

message file1_schema {
  OPTIONAL INT32 col_a;
  OPTIONAL BYTE_ARRAY col_b (UTF8);
}

message file2_schema {
  OPTIONAL BYTE_ARRAY col_b (UTF8);
  OPTIONAL BYTE_ARRAY col_c (UTF8);
}

EXTRACT
  FORMAT parquet
  SCHEMA (INFER_FROM sample_file)

EXTRACT
  FORMAT parquet
  SCHEMA (INFER_FROM sample_file)

CREATE PIPELINE testpipeline
    SOURCE S3
        BUCKET 'testbucket'
        ENDPOINT 'https://endpoint'
        ACCESS_KEY_ID ''
        SECRET_ACCESS_KEY ''
        FILTER_GLOB '/data/*/2024/*/*.parquet'
        PREFIX '/data/orders/2024/11/'
    EXTRACT
        FORMAT PARQUET
INTO parquet_base_table
SELECT $int32_field AS int32_field,
    $utf8_field as utf8_field,
    $float_field as float_field,
    $double_field as double_field,
    $int64_field as int64_field,
    $json_field as json_field,
    $bson_field as bson_field;

s3://data/orders/2024/11/dt=2024-11-24/file.parquet
.../dt=2024-11-25/file.parquet
.../dt=2024-11-26/file.parquet
...

SELECT int32_field,
    utf8_field,
    float_field,
    double_field,
    int64_field,
    json_field,
    bson_field,
    METADATA('hive_partition','dt') AS file_date;

<root>
  <person>
    <name>Barbara Smith</name>
    <address>
      <city>New York</city>
      <zip>12345</zip>
    </address>
    <note><![CDATA[Personal IP <b>127.0.0.1</b>]]></note>
  </person>
</root>

CREATE TABLE example_xml(
    name VARCHAR NOT NULL,
    city VARCHAR NOT NULL,
    zip VARCHAR NOT NULL,
    personal_ip IPV4 NOT NULL
);

CREATE PIPELINE xml_pipeline
    SOURCE S3
        BUCKET 'testbucket'
        ENDPOINT 'https://endpoint'
        ACCESS_KEY_ID ''
        SECRET_ACCESS_KEY ''
        FILTER_GLOB '/data/text.xml'
    EXTRACT
        FORMAT XML
INTO example_xml
SELECT
    $"root"."person"."name" as name,
    $"root"."address"."city" as city,
    $"root"."address"."zip" as zip,
    IPV4(SUBSTRING($"root"."person"."note", 16, 9)) as personal_ip;