<a id="flink-sql-confluent-cli"></a>

# Confluent CLI commands with Confluent Cloud for Apache Flink

Manage Flink SQL statements and compute pools in Confluent Cloud for Apache Flink® by using the
[confluent flink](https://docs.confluent.io/confluent-cli/current/command-reference/flink/index.html) 
commands in the Confluent CLI. To see the available commands, use the
`--help` option.

```bash
confluent flink statement --help
confluent flink compute-pool --help
confluent flink region --help
```

Use the Confluent CLI to manage these features:

- [Statements](#flink-cli-manage-statements)
- [Compute pools](#flink-cli-manage-compute-pools)
- [Regions](#flink-cli-list-regions)

For the complete CLI reference, see
[confluent flink statement](https://docs.confluent.io/confluent-cli/current/command-reference/flink/index.html).

In addition to the CLI, you can manage Flink statements and compute pools by
using these Confluent tools:

- [Flink SQL REST API](../operate-and-deploy/flink-rest-api.md#flink-rest-api)
- [Confluent Cloud Console](../get-started/quick-start-cloud-console.md#flink-sql-quick-start-run-sql-statement)
- [SQL shell](../get-started/quick-start-shell.md#flink-sql-quick-start-shell)
- [Confluent Terraform Provider](https://registry.terraform.io/providers/confluentinc/confluent/latest/docs)

<a id="flink-cli-manage-statements"></a>

## Manage statements

Using the Confluent CLI, you can perform these actions:

- [Submit a statement](#flink-sql-confluent-cli-submit-statement)
- [List statements](#flink-sql-confluent-cli-list-statements)
- [Describe a statement](#flink-sql-confluent-cli-describe-statement)
- [List exceptions from a statement](#flink-sql-confluent-cli-list-exceptions)
- [Delete a statement](#flink-sql-confluent-cli-delete-statement)
- [Update a statement](#flink-sql-confluent-cli-update-statement)

Managing Flink SQL statements can require the following inputs,
depending on the command:

```bash
export STATEMENT_NAME="<statement-name>" # example: "user-filter"
export COMPUTE_POOL_ID="<compute-pool-id>" # example: "lfcp-8m03rm"
export CLUSTER_ID="<kafka-cluster-id>" # example: "lkc-a1b2c3"
export PRINCIPAL_ID="<principal-id>" # example: "sa-23kgz4" for a service account, or "u-aq1dr2" for a user account
export SQL_CODE="<sql-statement-text>" # example: "SELECT * FROM USERS;"
```

For the complete CLI reference, see
[confluent flink statement](https://docs.confluent.io/confluent-cli/current/command-reference/flink/statement/index.html).

<a id="flink-sql-confluent-cli-submit-statement"></a>

### Submit a statement

The
[confluent flink statement create](https://docs.confluent.io/confluent-cli/current/command-reference/flink/statement/confluent_flink_statement_create.html)
command submits a statement in your compute pool.

Run the following command to submit a Flink SQL statement in the current
compute pool with your user account.

```bash
confluent flink statement create --sql "${SQL_CODE}"
```

Your output should resemble:

```none
+---------------+------------------------------------------------------------+
| Creation Date | 2024-02-28 21:08:08.9749 +0000                             |
|               | UTC                                                        |
| Name          | cli-2024-02-28-130806-78dd77b5-16a9-40ab-9786-db95b9895eaa |
| Statement     | Select 1;                                                  |
| Compute Pool  | lfcp-8m09g0                                                |
| Status        | PENDING                                                    |
+---------------+------------------------------------------------------------+
```

For long-running statements, Confluent recommends submitting statements with
a service account instead of your user account.

The following command submits a Flink SQL statement for the specified
principal in the specified compute pool and Flink database (Kafka cluster).

```bash
confluent flink statement create ${STATEMENT_NAME} \
  --service-account ${PRINCIPAL_ID} \
  --sql "${SQL_CODE}" \
  --compute-pool ${COMPUTE_POOL_ID} \
  --database ${CLUSTER_ID}
```

<a id="flink-sql-confluent-cli-list-statements"></a>

### List statements

Run the
[confluent flink statement list](https://docs.confluent.io/confluent-cli/current/command-reference/flink/statement/confluent_flink_statement_list.html)
command to list all of the non-deleted statements in your environment.

```bash
confluent flink statement list
```

Your output should resemble:

```none
          Creation Date         |         Name         |           Statement            | Compute Pool |  Status   |         Status Detail
--------------------------------+----------------------+--------------------------------+--------------+-----------+---------------------------------
  2023-07-08 21:04:06 +0000 UTC | 4b1d3494-f0f7-460d-9 | INSERT INTO copytopic          | lfcp-r2j1x9  | RUNNING   |
                                |                      | SELECT symbol,price from       |              |           |
                                |                      | topic_datagen;                 |              |           |
  2023-07-08 21:07:04 +0000 UTC | 6c43b973-b3c6-4be8-9 | INSERT INTO copytopic          | lfcp-r2j1x9  | RUNNING   |
                                |                      | SELECT symbol,price from       |              |           |
                                |                      | topic_datagen;                 |              |           |
...
```

To list only the statements that you’ve created, get the context for your
current Confluent Cloud login session and provide the context with the `context`
option.

```bash
confluent context list
```

Your output should resemble:

```none
  Current |                          Name                          |    Platform     |            Credential
----------+--------------------------------------------------------+-----------------+------------------------------------
  *       |   login-<your-email-address>-https://confluent.cloud   | confluent.cloud | username-<your-email-address>
```

For convenience, save the context in an environment variable:

```bash
export MY_CONTEXT="login-<your-email-address>-https://confluent.cloud"
```

Run the
[confluent flink statement list](https://docs.confluent.io/confluent-cli/current/command-reference/flink/statement/confluent_flink_statement_list.html)
command with your context.

```bash
confluent flink statement list ${MY_CONTEXT}
```

Your output should resemble:

```none
          Creation Date          |                            Name                            | Statement | Compute Pool |  Status   | Status Detail
---------------------------------+------------------------------------------------------------+-----------+--------------+-----------+----------------
  2024-02-28 21:08:08.9749 +0000 | cli-2024-02-28-130806-78dd77b5-16a9-40ab-9786-db95b9895eaa | Select 1; | lfcp-8m09g0  | COMPLETED |
  UTC                            |                                                            |           |              |           |
...
```

To list only the statements in your compute pool, provide the compute pool ID
with the `--compute-pool` option.

```bash
confluent flink statement list --compute-pool ${COMPUTE_POOL_ID}
```

<a id="flink-sql-confluent-cli-describe-statement"></a>

### Describe a statement

Run the
[confluent flink statement describe](https://docs.confluent.io/confluent-cli/current/command-reference/flink/statement/confluent_flink_statement_describe.html)
command to view the details of an existing statement.

```bash
confluent flink statement describe ${STATEMENT_NAME}
```

Your output should resemble:

```none
          Creation Date         |        Name        | Statement  | Compute Pool |  Status   | Status Detail
--------------------------------+--------------------+------------+--------------+-----------+----------------
  2023-07-19 19:26:52 +0000 UTC | fdc6cbf5-038a-408c | show jobs; | lfcp-a1b2c3  | COMPLETED |
```

<a id="flink-sql-confluent-cli-list-exceptions"></a>

### List exceptions from a statement

Run the
[confluent flink statement exception list](https://docs.confluent.io/confluent-cli/current/command-reference/flink/statement/exception/confluent_flink_statement_exception_list.html)
command to get exceptions that have been thrown by a statement.

```bash
confluent flink statement exception list ${STATEMENT_NAME}
```

<a id="flink-sql-confluent-cli-delete-statement"></a>

### Delete a statement

Run the
[confluent flink statement delete](https://docs.confluent.io/confluent-cli/current/command-reference/flink/statement/confluent_flink_statement_delete.html)
command to delete an existing statement permanently.

- All of its resources, like checkpoints, are also deleted.
- Deleting a statement stops charges for its use.

```bash
confluent flink statement delete ${STATEMENT_NAME}
```

Your output should resemble:

```none
Deleted Flink SQL statement "ac23db14-b5dc-49fb-b".
```

<a id="flink-sql-confluent-cli-update-statement"></a>

### Update a statement

Run the
[confluent flink statement update](https://docs.confluent.io/confluent-cli/current/command-reference/flink/statement/confluent_flink_statement_update.html)
command to stop an existing statement or resume a stopped statement. You can
also use this command to move a stopped statement to a different compute
pool or a different security principal when you resume it.

```bash
# Request to stop a statement.
confluent flink statement update ${STATEMENT_NAME} --stopped=true

# Request to resume a stopped statement.
confluent flink statement update ${STATEMENT_NAME} --stopped=false
```

To move a statement to a different compute pool, stop the statement, then
resume it with the `--compute-pool` option.

```bash
# Request to stop a statement.
confluent flink statement update ${STATEMENT_NAME} --stopped=true

# Request to resume the statement under a different compute pool.
confluent flink statement update ${STATEMENT_NAME} --stopped=false --compute-pool ${COMPUTE_POOL_ID}
```

You can combine `--compute-pool` with `--principal` to move a statement to
a different compute pool and run it as a different security principal at the
same time.

```bash
confluent flink statement update ${STATEMENT_NAME} --stopped=false --compute-pool ${COMPUTE_POOL_ID} --principal ${PRINCIPAL_ID}
```

<a id="flink-cli-manage-compute-pools"></a>

## Manage compute pools

Using the Confluent CLI, you can perform these actions:

- [Create a compute pool](#flink-sql-confluent-cli-create-compute-pool)
- [Describe a compute pool](#flink-sql-confluent-cli-describe-compute-pool)
- [List compute pools](#flink-sql-confluent-cli-list-compute-pools)
- [Update a compute pool](#flink-sql-confluent-cli-update-compute-pool)
- [Set the current compute pool](#flink-sql-confluent-cli-use-compute-pool)
- [Unset the current compute pool](#flink-sql-confluent-cli-unset-compute-pool)
- [Delete a compute pool](#flink-sql-confluent-cli-delete-compute-pool)

You must be authorized to create, update, delete (`FlinkAdmin`) or use
(`FlinkDeveloper`) a compute pool. For more information, see [Grant Role-Based Access in Confluent Cloud for Apache Flink](../operate-and-deploy/flink-rbac.md#flink-rbac).

Managing compute pools can require the following inputs, depending on the
command:

```bash
export COMPUTE_POOL_NAME=<compute-pool-name> # human-readable name, for example, "my-compute-pool"
export COMPUTE_POOL_ID="<compute-pool-id>" # example: "lfcp-8m03rm"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export MAX_CFU="<max-cfu>" # example: 5
```

For the complete CLI reference, see
[confluent flink compute-pool](https://docs.confluent.io/confluent-cli/current/command-reference/flink/compute-pool/index.html).

<a id="flink-sql-confluent-cli-create-compute-pool"></a>

### Create a compute pool

Run the
[confluent flink compute-pool create](https://docs.confluent.io/confluent-cli/current/command-reference/flink/compute-pool/confluent_flink_compute-pool_create.html)
command to create a compute pool.

Creating a compute pool requires the following inputs:

```bash
export COMPUTE_POOL_NAME=<compute-pool-name> # human-readable name, for example, "my-compute-pool"
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export MAX_CFU="<max-cfu>" # example: 5
```

Run the following command to create a compute pool in the specified cloud
provider and environment.

```bash
confluent flink compute-pool create ${COMPUTE_POOL_NAME} \
  --cloud ${CLOUD_PROVIDER} \
  --region ${CLOUD_REGION} \
  --max-cfu ${MAX_CFU} \
  --environment ${ENV_ID}
```

Your output should resemble:

```none
+-------------+-----------------+
| Current     | false           |
| ID          | lfcp-xxd6og     |
| Name        | my-compute-pool |
| Environment | env-z3y2x1      |
| Current CFU | 0               |
| Max CFU     | 5               |
| Cloud       | AWS             |
| Region      | us-east-1       |
| Status      | PROVISIONING    |
+-------------+-----------------+
```

#### NOTE
To create a default compute pool manually, add the `--default-pool` flag to
the command. This pre-provisions a default pool with your specified configuration
before users start running statements. The `FlinkAdmin` role or higher is
required to create default pools.

```bash
confluent flink compute-pool create ${COMPUTE_POOL_NAME} \
  --cloud ${CLOUD_PROVIDER} \
  --region ${CLOUD_REGION} \
  --max-cfu ${MAX_CFU} \
  --environment ${ENV_ID} \
  --default-pool
```

<a id="flink-sql-confluent-cli-describe-compute-pool"></a>

### Describe a compute pool

Run the
[confluent flink compute-pool describe](https://docs.confluent.io/confluent-cli/current/command-reference/flink/compute-pool/confluent_flink_compute-pool_describe.html)
command to get details about a compute pool.

Describing a compute pool requires the following inputs:

```bash
export COMPUTE_POOL_ID="<compute-pool-id>" # example: "lfcp-8m03rm"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
```

Run the following command to get details about a compute pool in the specified
environment.

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

Your output should resemble:

```none
+-------------+-----------------+
| Current     | false           |
| ID          | lfcp-xxd6og     |
| Name        | my-compute-pool |
| Environment | env-z3y2x1      |
| Current CFU | 0               |
| Max CFU     | 5               |
| Cloud       | AWS             |
| Region      | us-east-1       |
| Status      | PROVISIONED     |
+-------------+-----------------+
```

<a id="flink-sql-confluent-cli-list-compute-pools"></a>

### List compute pools

Run the
[confluent flink compute-pool list](https://docs.confluent.io/confluent-cli/current/command-reference/flink/compute-pool/confluent_flink_compute-pool_list.html)
command to compute pools in the specified environment.

Listing compute pools can require the following inputs, depending on the
command:

```bash
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
```

Run the following command to get details about a compute pool in the specified
environment.

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

Your output should resemble:

```none
  Current |     ID      |           Name            | Environment | Current CFU | Max CFU | Cloud |  Region   |   Status
----------+-------------+---------------------------+-------------+-------------+---------+-------+-----------+--------------
  *       | lfcp-xxd6og | my-compute-pool           | env-z3y2x1  |           0 |       5 | AWS   | us-east-1 | PROVISIONED
          | lfcp-8m03rm | test-blue-compute-pool    | env-z3q9rd  |           0 |      10 | AWS   | us-east-1 | PROVISIONED
...
```

<a id="flink-sql-confluent-cli-update-compute-pool"></a>

### Update a compute pool

Run the
[confluent flink compute-pool update](https://docs.confluent.io/confluent-cli/current/command-reference/flink/compute-pool/confluent_flink_compute-pool_update.html)
command to update a compute pool.

Updating a compute pool can require the following inputs, depending on the
command:

```bash
export COMPUTE_POOL_NAME=<compute-pool-name> # human-readable name, for example, "my-compute-pool"
export COMPUTE_POOL_ID="<compute-pool-id>" # example: "lfcp-8m03rm"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
export MAX_CFU="<max-cfu>" # example: 5
```

Run the following command to update a compute pool in the specified
environment.

```bash
confluent flink compute-pool update ${COMPUTE_POOL_ID} \
  --environment ${ENV_ID} \
  --name ${COMPUTE_POOL_NAME} \
  --max-cfu ${MAX_CFU}
```

Your output should resemble:

```none
+-------------+----------------------+
| Current     | false                |
| ID          | lfcp-xxd6og          |
| Name        | renamed-compute-pool |
| Environment | env-z3y2x1           |
| Current CFU | 0                    |
| Max CFU     | 10                   |
| Cloud       | AWS                  |
| Region      | us-east-1            |
| Status      | PROVISIONED          |
+-------------+----------------------+
```

<a id="flink-sql-confluent-cli-use-compute-pool"></a>

### Set the current compute pool

Run the
[confluent flink compute-pool use](https://docs.confluent.io/confluent-cli/current/command-reference/flink/compute-pool/confluent_flink_compute-pool_use.html)
command to use a compute pool in subsequent commands.

Setting a compute pool requires the following inputs:

```bash
export COMPUTE_POOL_ID="<compute-pool-id>" # example: "lfcp-8m03rm"
export ENV_ID="<environment-id>" # example: "env-z3y2x1"
```

Run the following commands to set the current compute pool in the specified
environment. First, you must run the `confluent environment use` command to
set the current environment.

```bash
confluent environment use ${ENV_ID} && \
confluent flink compute-pool use ${COMPUTE_POOL_ID}
```

Your output should resemble:

```none
Using environment "env-z3y2x1".
Using Flink compute pool "lfcp-xxd6og".
```

<a id="flink-sql-confluent-cli-unset-compute-pool"></a>

### Unset the current compute pool

Run the
[confluent flink compute-pool unset](https://docs.confluent.io/confluent-cli/current/command-reference/flink/compute-pool/confluent_flink_compute-pool_unset.html)
command to unset the current compute pool.

Run the following command to unset the current compute pool.

```bash
confluent flink compute-pool unset
```

Your output should resemble:

```none
Unset Flink compute pool "lfcp-xxd6og".
```

<a id="flink-sql-confluent-cli-delete-compute-pool"></a>

### Delete a compute pool

Run the
[confluent flink compute-pool delete](https://docs.confluent.io/confluent-cli/current/command-reference/flink/compute-pool/confluent_flink_compute-pool_delete.html)
command to delete a compute pool.

Run the following command to delete a compute pool in the specified
environment. The optional `--force` flag skips the confirmation prompt.

```bash
confluent flink compute-pool delete ${COMPUTE_POOL_ID} \
  --environment ${ENV_ID}
  --force
```

Your output should resemble:

```none
Deleted Flink compute pool "lfcp-xxd6og".
```

<a id="flink-cli-manage-regions"></a>

## Manage regions

Using the Confluent CLI, you can perform these actions:

- [List available regions](#flink-cli-list-regions)
- [Set the current region](#flink-cli-use-region)

Managing Flink SQL regions can require the following inputs,
depending on the command:

```bash
export CLOUD_PROVIDER="<cloud-provider>" # example: "aws"
export CLOUD_REGION="<cloud-region>" # example: "us-east-1"
```

For the complete CLI reference, see
[confluent flink region](https://docs.confluent.io/confluent-cli/current/command-reference/flink/region/index.html).

<a id="flink-cli-list-regions"></a>

### List available regions

Run the
[confluent flink region list](https://docs.confluent.io/confluent-cli/current/command-reference/flink/region/confluent_flink_region_list.html)
to see all available regions where you can run Flink statements.

```bash
confluent flink region list
```

Your output should resemble:

```none
  Current |             Name              | Cloud |        Region
----------+-------------------------------+-------+-----------------------
          | Belgium (europe-west1)        | gcp   | europe-west1
          | Frankfurt (eu-central-1)      | aws   | eu-central-1
          | Frankfurt (europe-west3)      | gcp   | europe-west3
          | Iowa (us-central1)            | gcp   | us-central1
          | Ireland (eu-west-1)           | aws   | eu-west-1
          | Las Vegas (us-west4)          | gcp   | us-west4
          | London (eu-west-2)            | aws   | eu-west-2
  *       | N. Virginia (us-east-1)       | aws   | us-east-1
          | N. Virginia (us-east4)        | gcp   | us-east4
          | Netherlands (westeurope)      | azure | westeurope
          | Ohio (us-east-2)              | aws   | us-east-2
          | Oregon (us-west-2)            | aws   | us-west-2
          | S. Carolina (us-east1)        | gcp   | us-east1
          | Singapore (ap-southeast-1)    | aws   | ap-southeast-1
          | Singapore (asia-southeast1)   | gcp   | asia-southeast1
          | Singapore (southeastasia)     | azure | southeastasia
          | Sydney (ap-southeast-2)       | aws   | ap-southeast-2
          | Sydney (australia-southeast1) | gcp   | australia-southeast1
          | Virginia (eastus)             | azure | eastus
          | Virginia (eastus2)            | azure | eastus2
          | Washington (westus2)          | azure | westus2
```

Run the following command to filter the list of available regions by cloud
provider.

```bash
confluent flink region list --cloud ${CLOUD_PROVIDER}
```

Your output should resemble:

```none
  Current |            Name            | Cloud |     Region
----------+----------------------------+-------+-----------------
          | Frankfurt (eu-central-1)   | aws   | eu-central-1
          | Ireland (eu-west-1)        | aws   | eu-west-1
          | London (eu-west-2)         | aws   | eu-west-2
  *       | N. Virginia (us-east-1)    | aws   | us-east-1
          | Ohio (us-east-2)           | aws   | us-east-2
          | Oregon (us-west-2)         | aws   | us-west-2
          | Singapore (ap-southeast-1) | aws   | ap-southeast-1
          | Sydney (ap-southeast-2)    | aws   | ap-southeast-2
```

<a id="flink-cli-use-region"></a>

### Set the current region

Run the
[confluent flink region use](https://docs.confluent.io/confluent-cli/current/command-reference/flink/region/confluent_flink_region_use.html)
to set the current region where subsequent Flink statements run. You must have
a compute pool in the region to run statements.

```bash
confluent flink region use --cloud ${CLOUD_PROVIDER} --region ${CLOUD_REGION}
```

For `CLOUD_PROVIDER=aws` and `CLOUD_REGION=us-east-2`, your output should
resemble:

```none
Using Flink region "Ohio (us-east-2)".
```

## Related content

- [Flink SQL Shell Quick Start](../get-started/quick-start-shell.md#flink-sql-quick-start-shell)
- [Monitor Flink SQL Statements](../operate-and-deploy/monitor-statements.md#flink-sql-monitor-statements-with-cloud-console)
- [Flink SQL REST API](../operate-and-deploy/flink-rest-api.md#flink-rest-api)
- [Confluent Terraform Provider](https://registry.terraform.io/providers/confluentinc/confluent/latest/docs)

#### 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).