Create Table SQL (DDLs)¶

This chapter offers a compilation of best practices for implementing Flink SQL solutions, applicable to local Flink open-source, the Confluent Platform for Flink or the Confluent Cloud for Flink.

Data Definition Language (DDL) are statements to define metadata in Flink SQL by creating, updating, or deleting tables. See the Flink SQL Examples in Confluent Cloud for Apache Flink documentation.

A table registered with the CREATE TABLE statement can be used as a table source or a table sink.

Create table statements do not changes between managed services and standalone Flink deployment, except the metadata. Each examples with specific standalone content, will be marked as Flink OSS, which also works for Confluent Platform Flink.

Sources Of Information¶

• Confluent SQL documentation for DDL with useful examples.
• Apache Flink CREATE SQL
• Confluent Developer Flink tutorials
• See SQL examples from product documentation.

Create table syntax

Common Principles¶

• Primary key can have one or more columns, all of them should be not null, and only being NOT ENFORCED
• Inside Flink processing, the primary key declaration, partitions the table implicitly by the key column(s)
• Flink uses the primary key for state management and deduplication with upsert. While the partition key is what determines which Kafka partition a message will be written to. This is a Kafka-level concept.

• For upsert mode, the bucket key must be equal to primary key. While for append/retract mode, the bucket key can be a subset of the primary key. See more on changelog.mode

• Understanding the table structure and mapping from a Kafka topic: the basic query helps a lot to understand table configuration and column types. This is helpful to assess time column, to confirm if the type of a column as event_time is a STRING, TIMESTAMP or TIMESTAMP_LTZ.

  show create table
  

  The TYPEOF(column_name) can also being used in a select to understand the type, and troubleshooting a type error.

• Understand the execution plan of a query using EXPLAIN insert or select, helps to assess if the query supports the expected semantic like window aggregation, interval joins or classical joins. The change to append to retract or upsert is also very important for state assessment. Recall that properties of the destination table will influence the plan.

Table Creation How-Tos¶

Confluent Cloud Specifics¶

• In Confluent Cloud, partition key will generate a key schema, except if using the option ('key.format' = 'raw')
• If the destination topic doesn't define partitioning key, then CC Flink SQL will write the records in whatever partitioning that was used at the end of the query. if last operation in the query is GROUP BY foo or JOIN ON A.foo = B.foo, then output records would be partitioned on foo values, and they wouldn't be re-partitioned before writing them into Kafka. The foo partitioning is preserved.

-- simplest table
CREATE TABLE humans (race STRING, origin STRING);
-- with primary key 
CREATE TABLE manufactures (m_id INT PRIMARY KEY NOT ENFORCED, site_name STRING);
-- with hash distribution to 2 partitions that match the primary key
CREATE TABLE humans (hid INT PRIMARY KEY NOT ENFORCED, race STRING, gender INT) DISTRIBUTED BY (hid) INTO 2 BUCKETS;

Flink OSS¶

create table user (
    'user_id' VARCHAR(250),
    'name' VARCHAR(50)
) partitioned by ('used-id')
WITH (
    'format' = 'json', -- other format are: csv, parquet
    'connector' = 'filesystem',
    'path' = '/tmp/users'
);

select .... from TableName 
WITH (
    'connector' = 'kafka',
    'topic' = 'flight_schedules',
    'properties.bootstrap.servers' = 'localhost:9092',
    'properties.group.id' = 'fs_grp',
    'scan.startup.mode' = 'earliest-offset',
)

CREATE TABLE shoe_customers (
    id STRING,
    first_name STRING,
    last_name STRING,
    email STRING,
    phone STRING,
    street_address STRING,
    state STRING,
    zip_code STRING,
    country STRING,
    country_code STRING
) WITH (
    'connector' = 'kafka',
    'topic' = 'shoe_customers',
    'properties.bootstrap.servers' = 'broker:29092',
    'scan.startup.mode' = 'earliest-offset',
    'key.format' = 'raw',
    'key.fields' = 'id',
    'value.format' = 'avro-confluent',
    'properties.group.id' = 'flink-sql-consumer',
    'value.fields-include' = 'ALL',
    'value.avro-confluent.url' = 'http://schema-registry:8081'
);

SET execution.runtime-mode=BATCH;

CREATE TABLE employee_info (
    emp_id INT,
    name VARCHAR,
    dept_id INT
) WITH ( 
    'connector' = 'filesystem',
    'path' = '/home/flink-sql-demos/00-basic-sql/data/employees.csv',
    'format' = 'csv'
);

SHOW TABLES;

SELECT * from employee_info WHERE dept_id = 101;

show create table orders;

alter table flight_schedules add(dt string metadata virtual);

create table shoe_customer_keyed(
    primary key(id) not enforced
) distributed by(id) into 1 buckets
as select id, first_name, last_name, email from shoe_customers;

-- DDL
create table t (
    StatesTable ROW< states ARRAY<ROW<name STRING, city STRING, lg DOUBLE, lat DOUBLE>>>,
    creationDate STRING
)

insert into t(group_id, StatesTable,creationDate)
values( 
    'grp_1',
    (
    ARRAY[ row('California', 'San Francisco', -122.4194, 37.7749),
        row('New York', 'New York City', -74.0060, 40.7128),
        row('Texas', 'Austin', -97.7431, 30.2672)
    ]
    ),            -- StatesTable
    '2020-10-10'  -- creationDate
)

select
group_id,
-- example of defining attribute from the idx element of an array (starts from 1) from a nested schema:
  CAST(StatesTable.states[3] AS DECIMAL(10, 4)) AS longitude,
  CAST(StatesTable.states[4] AS DECIMAL(10, 4)) AS latitude
from t
-- results
 'grp_1', -122.4194, 37.7749
 'grp_1', -74.0060, 40.7128
 'grp_1', -97.7431, 30.2672

CREATE TABLE `bounded_pageviews` (
  `url` STRING,
  `user_id` STRING,
  `browser` STRING,
  `ts` TIMESTAMP(3)
)
WITH (
  'connector' = 'faker',
  'number-of-rows' = '500',
  'rows-per-second' = '100',
  'fields.url.expression' = '/#{GreekPhilosopher.name}.html',
  'fields.user_id.expression' = '#{numerify ''user_##''}',
  'fields.browser.expression' = '#{Options.option ''chrome'', ''firefox'', ''safari'')}',
  'fields.ts.expression' =  '#{date.past ''5'',''1'',''SECONDS''}'
);

create table `some_clicks` (
      `order_id` STRING NOT NULL,
      ...
      `event_time` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'),
      PRIMARY KEY(order_id) NOT ENFORCED
distributed by hash(order_id) into '4' buckets

insert into `some_clicks`
select
    order_id, 
    user_id,
    $rowtime as event_time
from  `src_table`;

CREATE VIEW IF NOT EXISTS table_view AS SELECT ...

create table new_table (
    ....
    `event_time` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp',
 watermark for `event_time` as `event_time` - INTERVAL '30' SECOND
);

ALTER TABLE table_name MODIFY WATERMARK FOR $rowtime AS $rowtime - INTERVAL '1' SECOND;
-- in case we need to reverse back
ALTER TABLE table_name DROP WATERMARK;

create table `small-orders` (
    `order_id` STRING NOT NULL,
    `customer_id` INT NOT NULL,
    `product_id` STRING NOT NULL,
    `price` DOUBLE NOT NULL
) distributed by hash(order_id) into 1 buckets
with (
    'kafka.retention.time' = '1 d'
);

insert into `small-orders` select * from `examples`.`marketplace`.`orders` where price < 20;

ALTER TABLE <table_name> ADD (
    `headers` MAP<STRING,STRING> METADATA,
    `leader-epoch`INT METADATA VIRTUAL,
    `offset` BIGINT METADATA VIRTUAL,
    `partition` BIGINT METADATA VIRTUAL,
    `timestamp` TIMESTAMP_LTZ(3) METADATA,
    `timestamp-type` STRING METADATA VIRTUAL,
    `topic` STRING METADATA VIRTUAL
);

CREATE TABLE clicks_per_seconds (
    events_per_second BIGINT,
    window_time TIMESTAMP_LTZ(3) METADATA FROM 'timestamp'
    )

INSERT INTO clicks_per_seconds
SELECT
    COUNT(*) AS events_per_second,
    window_time
FROM TABLE(TUMBLE(TABLE clicks, DESCRIPTOR(`$rowtime`), INTERVAL '1' SECOND))
GROUP BY window_time, window_end, window_start

Applying to a Medallion Architecture¶

The current approach can be used for Flink pipeline processing:

Deeper dive¶

• Confluent product document - changelog
• Flink SQL Secrets: Mastering the Art of Changelog Event Out-of-Orderness
• Resolving: Primary key differs from derived upsert key

Confluent Cloud Flink table creation specific¶

See the product documentation with some specificities, like source and sink tables are directly mapped to Kafka Topics. The $rowtime TIMESTAMP_LTZ(3) NOT NULL is provided as a system column.

• For each topic there is an inferred table created. The catalog is the Confluent environment and the Kafka cluster is the database. We can use the ALTER TABLE statement to evolve schemas for those inferred tables.

• A table by default is mapped to a topic with 6 partitions, and the changelog being append. Primary key leads to an implicit DISTRIBUTED BY(k), and value and key schemas are created in Schema Registry. It is possible to create table with primary key and append mode, while by default it is a upsert mode.

  CREATE TABLE telemetries (
      device_id INT PRIMARY KEY NOT ENFORCED, 
      geolocation STRING, metric BIGINT,
      ts TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp')
  DISTRIBUTED INTO 4 BUCKETS
  WITH ('changelog.mode' = 'append');
  

The statement above also creates a metadata column for writing a Kafka message timestamp. This timestamp will not be defined in the schema registry. Compared to $rowtime which is declared as a METADATA VIRTUAL column, ts is selected in a SELECT * and is writable.

• When the primary key is specified, then it will not be part of the value schema, except if we specify (using value.fields-include' = 'all') that the value contains the full table schema. The payload of k is stored twice in Kafka message:

  CREATE TABLE telemetries (k INT, v STRING)
  DISTRIBUTED BY (k)
  WITH ('value.fields-include' = 'all');
  

• If the key is a string, it may make sense to do not have a schema for the key in this case declare (the key columns are determined by the DISTRIBUTED BY clause): This does not work if the key name is not key.

  CREATE TABLE telemetries (device_id STRING, metric BIGINT)
  DISTRIBUTED BY (key)
  WITH ('key.format' = 'raw');
  

• To keep the record in the topic forever add this kafka.retention.time' = '0' as options in the WITH. The supported units are:

  "d", "day", "h", "hour", "m", "min", "minute", "ms", "milli", "millisecond",
  "micro", "microsecond", "ns", "nano", "nanosecond"
  

  "fields": [
        {
        "name": "customer_id",
        "type": "string"
        },
        {
        "default": null,
        "name": "first_name",
        "type": [
            "null",
            "string"
        ]
        }
  ]

{
  "default": null,
  "name": "registration_date",
  "type": [
    "null",
    {
      "logicalType": "local-timestamp-millis",
      "type": "long"
    }
  ]
}

Analyzing Table¶

Understand the type of attribute or get the table structure with metadata¶

show create table 'tablename';
-- for a specific attribute
select typeof(column_name) from table_name limit 1;

Flink SQL planner performs type checking. Assessing the type of inferred table is helpful, specially around timestamp. See Data type mapping documentation.