<commit_before># Syntax Reference

| [Overview](/docs/) |[Quick Start](/docs/quickstart#quick-start) | [Concepts](/docs/concepts.md#concepts) | Syntax Reference | [Examples](/docs/examples.md#examples) | [FAQ](/docs/faq.md#frequently-asked-questions)  | [Roadmap](/docs/roadmap.md#roadmap) | [Demo](/docs/demo.md#demo) |
|---|----|-----|----|----|----|----|----|


The KSQL CLI provides a terminal-based interactive shell for running queries.

**Table of Contents**

- [CLI-specific commands](#cli-specific-commands)
- [KSQL statements](#ksql-commands)
  - [Scalar functions](#scalar-functions)
  - [Aggregate functions](#aggregate-functions)


# CLI-specific commands

These commands are non-KSQL statements such as setting a property or adding a resource. Run the CLI with the --help option to see the available options.

```
Description:
  The KSQL CLI provides a terminal-based interactive shell for running queries. Each command must be on a separate line. For KSQL command syntax, see the documentation at https://github.com/confluentinc/ksql/docs/.

help:
  Show this message.

clear:
  Clear the current terminal.

output:
  View the current output format.

output <format>:
  Set the output format to <format> (valid formats: 'JSON', 'TABULAR')
  For example: "output JSON"

history:
  Show previous lines entered during the current CLI session. You can use up and down arrow keys to navigate to the previous lines too.

version:
  Get the current KSQL version.

exit:
  Exit the CLI.


Default behavior:

    Lines are read one at a time and are sent to the server as KSQL unless one of the following is true:

    1. The line is empty or entirely whitespace. In this case, no request is made to the server.

    2. The line ends with backslash (`\`). In this case, lines are continuously read and stripped of their trailing newline and `\` until one is encountered that does not end with `\`; then, the concatenation of all lines read during this time is sent to the server as KSQL.
```

**Tip:** You can search and browse your bash history in the KSQL CLI with `CTRL + R`.  After pressing `CTRL + R`, start typing the command or any part of the command and an auto-complete of a past commands is shown.


# KSQL statements

KSQL statements should be terminated with a semicolon (`;`). If desired, in CLI use a back-slash
(`\`) to indicate continuation on the next line.


### DESCRIBE
```
DESCRIBE (stream_name|table_name);
```
List the columns in a stream or table along with their data type and other attributes.

### CREATE STREAM
```
CREATE STREAM stream_name (  { column_name data_type} [, ...] ) WITH ( property_name = expression [, ...] );
```
Create a new stream with the specified columns and properties.

The supported column data types are BOOELAN, INTEGER, BIGINT, DOUBLE, VARCHAR (STRING), ARRAY<ArrayType> (JSON only) and MAP<VARCHAR, ValueType> (JSON only).

In addition to the defined columns in the statement, KSQL adds two implicit columns to every stream, ROWKEY and ROWTIME, which represent the corresponding Kafka message key and message timestamp.

The possible properties to set in the WITH clause:
* KAFKA_TOPIC: The name of the Kafka topic that this streams is built upon. The topic should already exist in Kafka. This is a required property.
* VALUE_FORMAT: Specifies the format in which the value in the topic that data is serialized in. Currently, KSQL supports JSON, delimited. This is a required property.
* KEY: The name of the column that is the key.
* TIMESTAMP: The name of the column that will be used as the timestamp. This can be used to define the event time.

Example:

```sql
 CREATE STREAM pageview (viewtime bigint, userid varchar, pageid varchar) WITH (value_format = 'json', kafka_topic='pageview_topic_json');
```

### CREATE TABLE

```
CREATE TABLE table_name (  { column_name data_type} [, ...] ) WITH ( property_name = expression [, ...] );
```

Create a new KSQL table with the specified columns and properties. The supported column data types are BOOELAN, INTEGER, BIGINT, DOUBLE, VARCHAR (STRING), ARRAY<ArrayType> (JSON only) and MAP<VARCHAR, ValueType> (JSON only).

In addition to the defined columns in the statement, KSQL adds two implicit columns to every table, ROWKEY and ROWTIME, which represent the corresponding Kafka message key and message timestamp.

The possible properties to set in the WITH clause:
* KAFKA_TOPIC: The name of the Kafka topic that this streams is built upon. The topic should already exist in Kafka. This is a required property.
* VALUE_FORMAT: Specifies the format in which the value in the topic that data is serialized in. Currently, KSQL supports JSON, delimited. This is a required property.
* KEY: The name of the column that is the key.
* TIMESTAMP: The name of the column that will be used as the timestamp.

Example:

```sql
 CREATE TABLE users (usertimestamp bigint, userid varchar, gender varchar, regionid varchar) WITH (value_format = 'json', kafka_topic='user_topic_json');
```


### CREATE STREAM AS SELECT

Create a new KSQL stream along with the corresponding Kafka topic and stream the result of the SELECT query into the topic.

```sql
CREATE STREAM `stream_name`
[WITH ( `property_name = expression` [, ...] )]
AS SELECT  `select_expr` [, ...]
FROM `from_item` [, ...]
[ WHERE `condition` ]
[PARTITION BY `column_name`]
```

You can use the WITH section to set the properties for the result KSQL stream. The properties that
 can be set are:

* KAFKA_TOPIC: The name of KSQL topic and the corresponding Kafka topic associated with the new KSQL stream. If not specified, the name of the stream will be used as default.

* FORMAT: Specifies the format in which the result topic data is serialized in. KSQL supports JSON and DELIMITED. If not set the same format of the input stream will be used.

* PARTITIONS: The number of partitions in the sink stream.

* REPLICATIONS: The replication factor for the sink stream.

* TIMESTAMP: The name of the column that will be used as the timestamp. This can be used to define the event time.



### CREATE TABLE AS SELECT

Create a new KSQL table along with the corresponding Kafka topic and stream the result of the
SELECT query as a changelog into the topic.

```
CREATE TABLE `stream_name`
[WITH ( `property_name = expression` [, ...] )]
AS SELECT  `select_expr` [, ...]
FROM `from_item` [, ...]
[ WINDOW `window_expression` ]
[ WHERE `condition` ]
[ GROUP BY `grouping expression` ]
[ HAVING `having_expression` ]
```

The WITH section can be used to set the properties for the result KSQL topic. The properties that can be set are as the following:

* KAFKA_TOPIC: The name of KSQL topic and the corresponding Kafka topic associated with the new KSQL stream. If not set, the name of the stream will be used as default.

* FORMAT: Specifies the format in which the result topic data is serialized in. KSQL supports JSON and DELIMITED. If not set the same format of the input stream will be used.

* PARTITIONS: The number of partitions in the sink stream.

* REPLICATIONS: The replication factor for the sink stream.


###  DROP STREAM <stream-name>

Drops an existing stream.


### DROP TABLE <table-name>

Drops an existing table.


### SELECT

Selects rows from a KSQL stream or table. The result of this statement will not be persisted in a
 kafka topic and will only be printed out in the console. To stop the continuous query in the CLI
  press Ctrl+C.

```
SELECT `select_expr` [, ...]
FROM `from_item` [, ...]
[ WINDOW `window_expression` ]
[ WHERE `condition` ]
[ GROUP BY `grouping expression` ]
[ HAVING `having_expression` ]
```

In the above statements `from_item` is one of the following:

- `table_name [ [ AS ] alias]`<commit_msg>Applied Joel's feedback on rewording.<commit_after># Syntax Reference

| [Overview](/docs/) |[Quick Start](/docs/quickstart#quick-start) | [Concepts](/docs/concepts.md#concepts) | Syntax Reference | [Examples](/docs/examples.md#examples) | [FAQ](/docs/faq.md#frequently-asked-questions)  | [Roadmap](/docs/roadmap.md#roadmap) | [Demo](/docs/demo.md#demo) |
|---|----|-----|----|----|----|----|----|


The KSQL CLI provides a terminal-based interactive shell for running queries.

**Table of Contents**

- [CLI-specific commands](#cli-specific-commands)
- [KSQL statements](#ksql-commands)
  - [Scalar functions](#scalar-functions)
  - [Aggregate functions](#aggregate-functions)


# CLI-specific commands

These commands are non-KSQL statements such as setting a property or adding a resource. Run the CLI with the --help option to see the available options.

```
Description:
  The KSQL CLI provides a terminal-based interactive shell for running queries. Each command must be on a separate line. For KSQL command syntax, see the documentation at https://github.com/confluentinc/ksql/docs/.

help:
  Show this message.

clear:
  Clear the current terminal.

output:
  View the current output format.

output <format>:
  Set the output format to <format> (valid formats: 'JSON', 'TABULAR')
  For example: "output JSON"

history:
  Show previous lines entered during the current CLI session. You can use up and down arrow keys to navigate to the previous lines too.

version:
  Get the current KSQL version.

exit:
  Exit the CLI.


Default behavior:

    Lines are read one at a time and are sent to the server as KSQL unless one of the following is true:

    1. The line is empty or entirely whitespace. In this case, no request is made to the server.

    2. The line ends with backslash (`\`). In this case, lines are continuously read and stripped of their trailing newline and `\` until one is encountered that does not end with `\`; then, the concatenation of all lines read during this time is sent to the server as KSQL.
```

**Tip:** You can search and browse your bash history in the KSQL CLI with `CTRL + R`.  After pressing `CTRL + R`, start typing the command or any part of the command and an auto-complete of a past commands is shown.


# KSQL statements

KSQL statements should be terminated with a semicolon (`;`). If desired, in the CLI use a back-slash
(`\`) to indicate continuation on the next line.


### DESCRIBE
```
DESCRIBE (stream_name|table_name);
```
List the columns in a stream or table along with their data type and other attributes.

### CREATE STREAM
```
CREATE STREAM stream_name (  { column_name data_type} [, ...] ) WITH ( property_name = expression [, ...] );
```
Create a new stream with the specified columns and properties.

The supported column data types are BOOELAN, INTEGER, BIGINT, DOUBLE, VARCHAR (STRING), ARRAY<ArrayType> (JSON only) and MAP<VARCHAR, ValueType> (JSON only).

KSQL adds the implicit columns ROWTIME and ROWKEY to every stream and table, which represent the
corresponding Kafka message timestamp and message key.

These are the supported WITH clause properties:
* KAFKA_TOPIC: The name of the Kafka topic that this streams is built upon. The topic should already exist in Kafka. This is a required property.
* VALUE_FORMAT: Specifies the format in which the value in the topic that data is serialized in. Currently, KSQL supports JSON, delimited. This is a required property.
* KEY: The name of the key column.
* TIMESTAMP: The name of the timestamp column. This can be used to define the event time.

Example:

```sql
 CREATE STREAM pageview (viewtime bigint, userid varchar, pageid varchar) WITH (value_format = 'json', kafka_topic='pageview_topic_json');
```

### CREATE TABLE

```
CREATE TABLE table_name (  { column_name data_type} [, ...] ) WITH ( property_name = expression [, ...] );
```

Create a new KSQL table with the specified columns and properties. The supported column data types are BOOELAN, INTEGER, BIGINT, DOUBLE, VARCHAR (STRING), ARRAY<ArrayType> (JSON only) and MAP<VARCHAR, ValueType> (JSON only).

In addition to the defined columns in the statement, KSQL adds two implicit columns to every table, ROWKEY and ROWTIME, which represent the corresponding Kafka message key and message timestamp.

The possible properties to set in the WITH clause:
* KAFKA_TOPIC: The name of the Kafka topic that this streams is built upon. The topic should already exist in Kafka. This is a required property.
* VALUE_FORMAT: Specifies the format in which the value in the topic that data is serialized in. Currently, KSQL supports JSON, delimited. This is a required property.
* KEY: The name of the key column.
* TIMESTAMP: The name of the timestamp column.

Example:

```sql
 CREATE TABLE users (usertimestamp bigint, userid varchar, gender varchar, regionid varchar) WITH (value_format = 'json', kafka_topic='user_topic_json');
```


### CREATE STREAM AS SELECT

Create a new KSQL stream along with the corresponding Kafka topic and stream the result of the SELECT query into the topic.

```sql
CREATE STREAM `stream_name`
[WITH ( `property_name = expression` [, ...] )]
AS SELECT  `select_expr` [, ...]
FROM `from_item` [, ...]
[ WHERE `condition` ]
[PARTITION BY `column_name`]
```

You can use the WITH section to set the properties for the result KSQL stream. The properties that
 can be set are:

* KAFKA_TOPIC: The name of KSQL topic and the corresponding Kafka topic associated with the new KSQL stream. If not specified, the name of the stream will be used as default.

* FORMAT: Specifies the format in which the result topic data is serialized in. KSQL supports JSON and DELIMITED. If not set the same format of the input stream will be used.

* PARTITIONS: The number of partitions in the sink stream.

* REPLICATIONS: The replication factor for the sink stream.

* TIMESTAMP: The name of the column that will be used as the timestamp. This can be used to define the event time.



### CREATE TABLE AS SELECT

Create a new KSQL table along with the corresponding Kafka topic and stream the result of the
SELECT query as a changelog into the topic.

```
CREATE TABLE `stream_name`
[WITH ( `property_name = expression` [, ...] )]
AS SELECT  `select_expr` [, ...]
FROM `from_item` [, ...]
[ WINDOW `window_expression` ]
[ WHERE `condition` ]
[ GROUP BY `grouping expression` ]
[ HAVING `having_expression` ]
```

The WITH section can be used to set the properties for the result KSQL topic. The properties that can be set are as the following:

* KAFKA_TOPIC: The name of KSQL topic and the corresponding Kafka topic associated with the new KSQL stream. If not set, the name of the stream will be used as default.

* FORMAT: Specifies the format in which the result topic data is serialized in. KSQL supports JSON and DELIMITED. If not set the same format of the input stream will be used.

* PARTITIONS: The number of partitions in the sink stream.

* REPLICATIONS: The replication factor for the sink stream.


###  DROP STREAM <stream-name>

Drops an existing stream.


### DROP TABLE <table-name>

Drops an existing table.


### SELECT

Selects rows from a KSQL stream or table. The result of this statement will not be persisted in a
 Kafka topic and will only be printed out in the console. To stop the continuous query in the CLI
  press Ctrl+C.

```
SELECT `select_expr` [, ...]
FROM `from_item` [, ...]
[ WINDOW `window_expression` ]
[ WHERE `condition` ]
[ GROUP BY `grouping expression` ]
[ HAVING `having_expression` ]
```

In the above statements `from_item` is one of the following:

- `table_name [ [ AS ] alias]`
{'fraction_non_alphanumeric': 0.08723473718576559, 'fraction_numerical': 0.0002611818478615736, 'mean_word_length': 4.084993359893758, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 19, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '84431', 'n_tokens_mistral': 4471, 'n_tokens_neox': 4232, 'n_words': 2129}