Create a KSQL Table

In KSQL, you create tables from Apache Kafka® topics, and you create tables of query results from other tables or streams.

  • Use the CREATE TABLE statement to create a table from a Kafka topic.
  • Use the CREATE TABLE AS SELECT statement to create a table with query results from an existing table or stream.

Note

Creating streams is similar to creating tables. For more information, see Create a KSQL Stream.

Create a Table from a Kafka Topic

Use the CREATE TABLE statement to create a table from an underlying Kafka topic. The Kafka topic must exist already in your Kafka cluster.

The following examples show how to create tables from a Kafka topic, named users. To see these examples in action, create the users topic by following the procedure in Writing Streaming Queries Against Apache Kafka® Using KSQL (Docker).

Create a Table with Selected Columns

The following example creates a table that has four columns from the users topic: registertime, userid, gender, and regionid. Also, the userid field is assigned as the table's KEY property.

Note

The KEY field is required for all KSQL tables. For more information, see Key Requirements.

KSQL can't infer the topic's data format, so you must provide the format of the values that are stored in the topic. In this example, the data format is JSON. Other options are Avro and DELIMITED.

In the KSQL CLI, paste the following CREATE TABLE statement:

CREATE TABLE users \
  (registertime BIGINT, \
   userid VARCHAR, \
   gender VARCHAR, \
   regionid VARCHAR) \
  WITH (KAFKA_TOPIC = 'users', \
        VALUE_FORMAT='JSON', \
        KEY = 'userid');

Your output should resemble:

 Message
---------------
 Table created
---------------

Inspect the table by using the SHOW TABLES and DESCRIBE statements:

SHOW TABLES;

Your output should resemble:

 Table Name | Kafka Topic | Format | Windowed
----------------------------------------------
 USERS      | users       | JSON   | false
----------------------------------------------

Get the schema for the table:

DESCRIBE users;

Your output should resemble:

Name                 : USERS
 Field        | Type
------------------------------------------
 ROWTIME      | BIGINT           (system)
 ROWKEY       | VARCHAR(STRING)  (system)
 REGISTERTIME | BIGINT
 USERID       | VARCHAR(STRING)
 GENDER       | VARCHAR(STRING)
 REGIONID     | VARCHAR(STRING)
------------------------------------------
For runtime statistics and query details run: DESCRIBE EXTENDED <Stream,Table>;

Create a continuous streaming query on the users table by using the SELECT statement:

SELECT * FROM users;

Your output should resemble:

1541439611069 | User_2 | 1498028899054 | User_2 | MALE | Region_1
1541439611320 | User_6 | 1505677113995 | User_6 | FEMALE | Region_7
1541439611396 | User_5 | 1491338621627 | User_5 | OTHER | Region_2
1541439611536 | User_9 | 1492621173463 | User_9 | FEMALE | Region_3
^CQuery terminated

Press CTRL+C to stop printing the query results.

The table values update continuously with the most recent records, because the underlying users topic receives new messages continuously.

Create a KSQL Table with Streaming Query Results

Use the CREATE TABLE AS SELECT statement to create a KSQL table that contains the results of a SELECT query from another table or stream.

CREATE TABLE AS SELECT creates a new KSQL table with a corresponding Kafka topic and streams the result of the SELECT query as a changelog into the topic. KSQL creates a persistent query that runs continuously until you terminate it explicitly.

To stream the result of a SELECT query into an existing table and its underlying topic, use the INSERT INTO statement.

The following KSQL statement creates a users_female table that contains results from a persistent query for users that have gender set to FEMALE:

CREATE TABLE users_female AS \
  SELECT userid, gender, regionid FROM users \
  WHERE gender='FEMALE';

Your output should resemble:

 Message
---------------------------
 Table created and running
---------------------------

Inspect the table by using the SHOW TABLES and PRINT statements:

SHOW TABLES;

Your output should resemble:

 Table Name   | Kafka Topic  | Format | Windowed
-------------------------------------------------
 USERS        | users        | JSON   | false
 USERS_FEMALE | USERS_FEMALE | JSON   | false
-------------------------------------------------

Print some rows in the table:

PRINT users_female;

Your output should resemble:

Format:JSON
{"ROWTIME":1541458112639,"ROWKEY":"User_5","USERID":"User_5","GENDER":"FEMALE","REGIONID":"Region_4"}
{"ROWTIME":1541458112857,"ROWKEY":"User_2","USERID":"User_2","GENDER":"FEMALE","REGIONID":"Region_7"}
{"ROWTIME":1541458112838,"ROWKEY":"User_9","USERID":"User_9","GENDER":"FEMALE","REGIONID":"Region_4"}
^CTopic printing ceased

Press CTRL+C to stop printing the table.

Note

The query continues to run after you stop printing the table.

Use the SHOW QUERIES statement to view the query that KSQL created for the users_female table:

SHOW QUERIES;

Your output should resemble:

 Query ID            | Kafka Topic  | Query String
-----------------------------------------------------------------------------------------------------------------------------------------
 CTAS_USERS_FEMALE_0 | USERS_FEMALE | CREATE TABLE users_female AS   SELECT userid, gender, regionid FROM users   WHERE gender='FEMALE';
-----------------------------------------------------------------------------------------------------------------------------------------
For detailed information on a Query run: EXPLAIN <Query ID>;

A persistent query that's created by the CREATE TABLE AS SELECT statement has the string CTAS in its ID, for example, CTAS_USERS_FEMALE_0.

Delete a KSQL Table

Use the DROP TABLE statement to delete a table. If you created the table by using CREATE TABLE AS SELECT, you must first terminate the corresponding persistent query.

Use the TERMINATE statement to stop the CTAS_USERS_FEMALE_0 query:

TERMINATE CTAS_USERS_FEMALE_0;

Your output should resemble:

 Message
-------------------
 Query terminated.
-------------------

Use the DROP TABLE statement to delete the users_female table:

DROP TABLE users_female;

Your output should resemble:

 Message
-----------------------------------
 Source USERS_FEMALE was dropped.
-----------------------------------