<a id="flink-sql-quick-start-shell"></a>

# Flink SQL Shell Quick Start on Confluent Cloud for Apache Flink

Submit your first Flink SQL statement from the Flink SQL shell, an
interactive CLI for authoring and running statements on Confluent Cloud for Apache Flink®. The
following steps log you in to Confluent Cloud, start the shell, and run a query
against streaming data.

- [Step 1: Log in to Confluent Cloud with the Confluent CLI](#flink-sql-quick-start-shell-login)
- [Step 2: Start the Flink SQL shell](#flink-sql-quick-start-shell-sql-shell)
- [Step 3: Submit a SQL statement](#flink-sql-quick-start-shell-submit-statement)
- [Step 4: Create and populate a table](#flink-sql-quick-start-shell-create-table)
- [Step 5: Query streaming data](#flink-sql-quick-start-shell-query-streaming-data)

## Prerequisites

You need the following prerequisites to use the Flink SQL shell.

- Access to Confluent Cloud.
- The organization ID and environment ID for your organization.
- The cloud provider and region where you run your Flink SQL statements.
  Default compute pools are scoped to an environment and a region, so you
  must specify both unless you name a compute pool.
- (Optional) A compute pool ID, if you want to use a specific compute pool instead
  of the default pool. For more information, see [Compute Pools](../concepts/compute-pools.md#flink-sql-compute-pools).
- The FlinkDeveloper role is granted by default to all users in an environment.
  To create compute pools manually for workload isolation or cost control,
  you need the OrganizationAdmin, EnvironmentAdmin, or FlinkAdmin role.
  If you don’t have the appropriate role, contact your OrganizationAdmin
  or EnvironmentAdmin.
- The Confluent CLI. To use the Flink SQL shell, update to the latest
  version of the Confluent CLI by running the following command:
  ```bash
  confluent update --yes
  ```

  If you used Homebrew to install the Confluent CLI, update the CLI by
  using the `brew upgrade` command, instead of `confluent update`.

  For more information, see [Confluent CLI](https://docs.confluent.io/confluent-cli/current/overview.html).

<a id="flink-sql-quick-start-shell-login"></a>

## Step 1: Log in to Confluent Cloud with the Confluent CLI

Run the following CLI command to log in to Confluent Cloud.

```bash
confluent login --save --organization ${ORG_ID}
```

Your output should resemble:

```none
Logged in as "<your-email>" for organization "<your-org-id>" ("<your-org-name>").
```

<a id="flink-sql-quick-start-shell-sql-shell"></a>

## Step 2: Start the Flink SQL shell

Start the Flink SQL shell by running the `confluent flink shell` command.
The shell connects with Confluent Cloud. By default, the Confluent CLI connects to
Flink over the public endpoint, even if your environment has private
networking set up, like a Confluent Cloud network (CCN) or PrivateLink Gateway. If your environment
uses private networking, select a private endpoint before you start the
shell. For more information, see [Access private networking with the Confluent CLI](../concepts/flink-private-networking.md#flink-sql-private-networking-cli).

#### IMPORTANT
This guide focuses on ad hoc statements. To run statements in long-running
jobs, provide the `--service-account` option in the
`confluent flink shell` command. When you start the shell
without this option, statements run with your user account. For more
information, see [Service Accounts on Confluent Cloud](../../security/authenticate/workload-identities/service-accounts/overview.md#service-accounts).

Run the following CLI command to start the Flink SQL shell.

```bash
confluent flink shell \
  --environment ${ENV_ID} \
  --cloud ${CLOUD} \
  --region ${REGION}
```

Your output should resemble:

```none
Welcome!
To exit, press Ctrl-Q or type "exit".

[Ctrl-Q] Quit [Ctrl-S] Toggle Smart Completion
>
```

The shell uses the default compute pool for the environment and region that
you specify. Default compute pools are scoped to an environment and a region,
so the `--cloud` and `--region` options are required unless you name a pool
with the `--compute-pool` option:

```bash
confluent flink shell --compute-pool ${COMPUTE_POOL_ID} --environment ${ENV_ID}
```

You’re ready to start processing data by submitting statements to Flink SQL.

<a id="flink-sql-quick-start-shell-submit-statement"></a>

## Step 3: Submit a SQL statement

In the SQL shell, run the following statement to see Flink SQL in action.
The [CURRENT_TIMESTAMP](../reference/functions/datetime-functions.md#flink-sql-current-timestamp-function) function returns the local
date and time.

```sql
SELECT CURRENT_TIMESTAMP;
```

Your output should resemble:

```none
Creating statement: cli-2026-02-10-123456-7a8b9c0d-1234-5a6b-7890-808d0bbfecc6
Statement successfully submitted.
Finished statement execution. Statement phase: RUNNING.
+-------------------------+
|    CURRENT_TIMESTAMP    |
+-------------------------+
| 2026-02-10 11:25:09.903 |
+-------------------------+
```

For all functions and statements supported by Flink SQL, see
[Flink SQL Reference](../reference/overview.md#flink-sql-reference).

<a id="flink-sql-quick-start-shell-create-table"></a>

## Step 4: Create and populate a table

The following steps show how to create a table, populate it with
a few records, and query it to view the records it contains.

1. Run the following statement to create a table that contains pseudorandom
   float values.
   ```sql
   CREATE TABLE random_float_table(
     ts TIMESTAMP_LTZ(3),
     random_value FLOAT);
   ```
2. Run the following [INSERT VALUES](../reference/queries/insert-values.md#flink-sql-insert-values-statement)
   statement to populate `random_float_table` with records that have a
   timestamp field and a `float` field. The
   [CURRENT_TIMESTAMP](../reference/functions/datetime-functions.md#flink-sql-current-timestamp-function) function generates the
   timestamp values, and the [RAND_INTEGER(INT)](../reference/functions/numeric-functions.md#flink-sql-rand-integer-function) function,
   multiplied by a float, generates the float values.
   ```sql
   INSERT INTO random_float_table VALUES
     (CURRENT_TIMESTAMP, RAND_INTEGER(100)*0.02),
     (CURRENT_TIMESTAMP, RAND_INTEGER(1000)*0.05),
     (CURRENT_TIMESTAMP, RAND_INTEGER(10000)*0.20),
     (CURRENT_TIMESTAMP, RAND_INTEGER(100000)*0.22),
     (CURRENT_TIMESTAMP, RAND_INTEGER(1000000)*0.7);
   ```

   Press ENTER to return to the SQL shell. Because INSERT INTO VALUES is a
   point-in-time statement, it exits after it completes inserting records.
3. Run the following statement to query `random_float_table` for all of its
   records.
   ```sql
   SELECT * FROM random_float_table;
   ```

   Your output should resemble:
   ```none
   ts                      random_value
   2026-02-10 20:24:19.366 0.46
   2026-02-10 20:24:19.276 28.75
   2026-02-10 20:24:19.367 1467.2
   2026-02-10 20:24:19.368 7953.88
   2026-02-10 20:24:19.465 685883.1
   ```

   Press Q to exit the results view and stop the statement.
4. Run the [SHOW JOBS](../reference/statements/show.md#flink-sql-show-jobs) statement to get the status of statements
   in your SQL environment.
   ```sql
   SHOW JOBS;
   ```

   Your output should resemble:
   ```none
   Creating statement: cli-2026-02-10-123456-7a8b9c0d-1234-5a6b-7890-808d0bbfecc6
   Statement successfully submitted.
   Finished statement execution. Statement phase: COMPLETED.
   +----------------------------------+-----------+------------------+--------------+------------------+------------------+
   |        Name                      |   Phase   |    Statement     | Compute Pool |  Creation Time   |      Detail      |
   +----------------------------------+-----------+------------------+--------------+------------------+------------------+
   | cli-2026-02-10-080946-155640b... | COMPLETED | CREATE TABLE ... | lfcp-xxxxxx  | 2026-02-10 20... | This statemen... |
   | cli-2026-02-10-123456-7a8b9c0... | COMPLETED | SELECT CURREN... | lfcp-xxxxxx  | 2026-02-10 20... |                  |
   +----------------------------------+-----------+------------------+--------------+------------------+------------------+
   ```

<a id="flink-sql-quick-start-shell-query-streaming-data"></a>

## Step 5: Query streaming data

Flink SQL enables using familiar SQL syntax to query streaming data.
Confluent Cloud for Apache Flink provides [example data streams](../reference/example-data.md#flink-sql-example-data)
that you can experiment with. In this step, you query the `orders`
table from the `marketplace` database in the `examples` catalog.

In Flink SQL, the catalog and database scope catalog objects, like tables.

- A *catalog* is a collection of databases that share the same namespace.
- A *database* is a collection of tables that share the same namespace.

In Confluent Cloud, an environment maps to a Flink catalog, and
a Kafka cluster maps to a Flink database.

You can always use three-part identifiers for your tables, like
`catalog.database.table`, but it’s more convenient to set a default.

1. Run the following statement to set the default catalog.
   ```sql
   USE CATALOG `examples`;
   ```

   Your output should resemble:
   ```none
   +---------------------+----------+
   |         Key         |  Value   |
   +---------------------+----------+
   | sql.current-catalog | examples |
   +---------------------+----------+
   ```
2. Run the following statement to set the default database.
   ```sql
   USE `marketplace`;
   ```

   Your output should resemble:
   ```none
   +----------------------+-------------+
   |         Key          |    Value    |
   +----------------------+-------------+
   | sql.current-database | marketplace |
   +----------------------+-------------+
   ```
3. Run the following statement to see the list of available tables.
   ```sql
   SHOW TABLES;
   ```

   Your output should resemble:
   ```none
   +------------+
   | Table Name |
   +------------+
   | clicks     |
   | customers  |
   | orders     |
   | products   |
   +------------+
   ```
4. Run the following statement to inspect the `orders` data stream.
   ```sql
   SELECT * FROM orders;
   ```

   Your output should resemble:
   ```none
   order_id                             customer_id product_id price
   36d77b21-e68f-4123-b87a-cc19ac1f36ac 3137        1305       65.71
   7fd3cd2a-392b-4f8f-b953-0bfa1d331354 3063        1327       17.75
   1a223c61-38a5-4b8c-8465-2a6b359bf05e 3064        1166       14.95
   ...
   ```
5. Press Q to exit the results view and stop the statement.

Congratulations, you have run your first Flink SQL statements on Confluent Cloud
using the SQL shell.

## Next steps

- [How-to Guides for Confluent Cloud for Apache Flink](../how-to-guides/overview.md#flink-sql-how-to-guides)

## Related content

- [Course: Apache Flink 101](https://developer.confluent.io/courses/apache-flink/intro/).
- [Course: Building Flink Applications in Java](https://developer.confluent.io/courses/flink-java/overview/).
- [DDL Statements](../concepts/statements.md#flink-sql-statements)
- [Stream Processing Concepts](../concepts/overview.md#flink-sql-stream-processing-concepts)
- [Built-in Functions](../reference/functions/overview.md#flink-sql-functions-overview)

#### NOTE
This website includes content developed at the [Apache Software Foundation](https://www.apache.org/)
under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).