<a id="flink-python-table-api-quick-start"></a>

# Python Table API Quick Start on Confluent Cloud for Apache Flink

Confluent Cloud for Apache Flink® supports programming applications with the Table API. Confluent
provides a plugin for running applications that use the Table API on Confluent Cloud.

For more information, see [Table API](../reference/table-api.md#flink-table-api).

For code examples, see
[Python Examples for Table API on Confluent Cloud](https://github.com/confluentinc/flink-table-api-python-examples).

#### NOTE
The Flink Table API for Python is available for preview. The Table API for
Java is generally available. For more information, see the
[Java Table API Quick Start](quick-start-java-table-api.md#flink-java-table-api-quick-start).

<!-- Admonition for product maturity stage: OP -->
<!-- Use this file for standard legalese in docs for new Open Preview features. -->
<!-- Suggested usage: -->
<!-- .. note: (add a second colon here) -->
<!-- Feature X is an Open Preview feature in |product|. -->
<!-- .. include:: ../.hidden/docs-common/home/includes/product-maturity-stage-admonition-op.rst -->
<!-- Reference: Product maturity stages and docs -->
<!-- https://confluentinc.atlassian.net/wiki/spaces/DOC/pages/2695004342/Product+maturity+stages+and+docs -->

A Preview feature is a Confluent Cloud component that is being introduced to gain
early feedback from developers. Preview features can be used for evaluation
and non-production testing purposes or to provide feedback to Confluent.
The warranty, SLA, and Support Services provisions of your agreement with
Confluent do not apply to Preview features. Confluent may discontinue
providing preview releases of the Preview features at any time in
Confluent’s’ sole discretion.

Submit comments, questions, and suggestions related to the Table API
through the [established channels](../get-help.md#ccloud-flink-help).

## Prerequisites

- Access to Confluent Cloud.
- A [compute pool](../operate-and-deploy/create-compute-pool.md#flink-sql-manage-compute-pool) in Confluent Cloud.
- A Apache Kafka® [cluster](../../clusters/create-cluster.md#cloud-create-cluster), if you want to run examples
  that store data in Kafka.
- Java version 17 or later.
- The [uv](https://docs.astral.sh/uv/) package manager to manage your Python
  versions and environments. Use Python versions 3.9 to 3.11 only.

To run Table API and Flink SQL programs, you must have an API key. Also,
you need Confluent Cloud account details, like your organization and environment
identifiers.

- **Global API Key:** The recommended option. A
  [global API key](../../security/authenticate/workload-identities/service-accounts/api-keys/overview.md#cloud-global-api-keys) works for both Flink access
  and artifact creation, for example, for UDF uploads. Create one by running
  the following command:
  ```bash
  confluent api-key create --resource global
  ```

  In Confluent Cloud Console, you can create the same key by going to the
  [API keys page](https://confluent.cloud/settings/api-keys), clicking
  **Add API key**, selecting your account, and choosing the **Global**
  resource scope. Assign the key and secret to the `CONFLUENT_GLOBAL_API_KEY`
  and `CONFLUENT_GLOBAL_API_SECRET` environment variables. For the full
  procedure, see [Add an API key](../../security/authenticate/workload-identities/service-accounts/api-keys/manage-api-keys.md#create-api-key).

  As an alternative, you can use an API key that’s specific to the Flink
  environment by following the steps in [Generate a Flink API key](../operate-and-deploy/flink-rest-api.md#flink-rest-api-generate-api-key)
  and assigning the key and secret to the `CONFLUENT_FLINK_API_KEY` and
  `CONFLUENT_FLINK_API_SECRET` environment variables.
- **Organization ID:** The identifier of your organization, for example,
  `b0b421724-4586-4a07-b787-d0bb5aacbf87`. For convenience, assign your
  organization identifier to the CONFLUENT_ORG_ID environment variable.
- **Environment ID:** The identifier of the environment where your Flink SQL
  statements run, for example, `env-z3y2x1`. For convenience, assign your
  environment identifier to the CONFLUENT_ENV_ID environment variable.
- **Cloud provider name:** The name of the cloud provider where your cluster
  runs, for example, `aws`. To see the available providers, run the
  `confluent flink region list` command. For convenience, assign your
  cloud provider to the CONFLUENT_CLOUD_PROVIDER environment variable.
- **Cloud region:** The name of the region where your cluster runs, for
  example, `us-east-1`. To see the available regions, run the
  `confluent flink region list` command. For convenience, assign your
  cloud region to the CONFLUENT_CLOUD_REGION environment variable.

```bash
export CONFLUENT_CLOUD_PROVIDER="aws"
export CONFLUENT_CLOUD_REGION="us-east-1"
export CONFLUENT_GLOBAL_API_KEY="<your-global-api-key>"
export CONFLUENT_GLOBAL_API_SECRET="<your-global-api-secret>"
export CONFLUENT_ORG_ID="<your-organization-id>"
export CONFLUENT_ENV_ID="<your-environment-id>"
export CONFLUENT_COMPUTE_POOL_ID="<your-compute-pool-id>"
```

#### NOTE
The Flink Python API communicates with a Java process. You must have at least
Java 17 installed. Confirm that your `JAVA_HOME` environment variable
points to a valid Java install. Checking only `java -version` might not
be sufficient.

```bash
echo $JAVA_HOME
```

If required, install openjdk and export the `JAVA_HOME` variable:

```bash
brew install openjdk && export JAVA_HOME=$(/usr/libexec/java_home) && echo $JAVA_HOME
```

## Set up your environment and run a Table API program

Use [uv](https://docs.astral.sh/uv/) to create a virtual environment
that contains all required dependencies and project files.

1. Use one of the following commands to install uv.
   ```bash
   curl -LsSf https://astral.sh/uv/install.sh | sh
   # or
   brew install uv
   # or
   pip install uv
   ```
2. Create a new virtual environment.
   ```bash
   uv venv --python 3.11
   ```
3. Copy the following code into a file named `hello_table_api.py`.
   ```python
   # /// script
   # requires-python = ">=3.9,<3.12"
   # dependencies = [
   #   "confluent-pyflink>=2.3.3",
   # ]
   # ///

   from confluent_pyflink.table.utils import ConfluentSettings, ConfluentTools
   from confluent_pyflink.table import TableEnvironment, Row
   from confluent_pyflink.table.expressions import col, row

   def run():
       # Set up the connection to Confluent Cloud
       settings = ConfluentSettings.from_global_variables()
       env = TableEnvironment.create(settings)

       # Run your first Flink statement in Table API
       env.from_elements([row("Hello world!")]).execute().print()

       # Or use SQL
       env.sql_query("SELECT 'Hello world!'").execute().print()

       # Structure your code with Table objects - the main ingredient of Table API.
       table = env.from_path("examples.marketplace.clicks") \
           .filter(col("user_agent").like("Mozilla%")) \
           .select(col("click_id"), col("user_id"))

       table.print_schema()
       print(table.explain())

       # Use the provided tools to test on a subset of the streaming data
       expected = ConfluentTools.collect_materialized_limit(table, 50)
       actual = [Row(42, 500)]
       if expected != actual:
           print("Results don't match!")

   if __name__ == "__main__":
       run()
   ```
4. Run the following command to execute the Table API program from the
   directory where you created `hello_table_api.py`.
   ```bash
   uv run hello_table_api.py
   ```

## Related content

- [Filter Kafka messages in Python using Flink’s Table API](https://developer.confluent.io/confluent-tutorials/filtering/flink_table_api_python/).
- GitHub repo: [Java Examples for Table API on Confluent Cloud](https://github.com/confluentinc/flink-table-api-java-examples).
- GitHub repo: [Python Examples for Table API on Confluent Cloud](https://github.com/confluentinc/flink-table-api-python-examples).
- [Java Table API Quick Start](quick-start-java-table-api.md#flink-java-table-api-quick-start)
- [How-to Guides for Confluent Cloud for Apache Flink](../how-to-guides/overview.md#flink-sql-how-to-guides)
- [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).