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.

For code examples, see Python Examples for Table API on Confluent Cloud.

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.

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.

Prerequisites

Access to Confluent Cloud.
A compute pool in Confluent Cloud.
A Apache Kafka® cluster, if you want to run examples that store data in Kafka.
Java version 17 or later.
The 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 works for both Flink access and artifact creation, for example, for UDF uploads. Create one by running the following command:
```
confluent api-key create --resource global
```
In Confluent Cloud Console, you can create the same key by going to the API keys page, 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.

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

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.

echo $JAVA_HOME

If required, install openjdk and export the JAVA_HOME variable:

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 to create a virtual environment that contains all required dependencies and project files.

Use one of the following commands to install uv.

curl -LsSf https://astral.sh/uv/install.sh | sh
# or
brew install uv
# or
pip install uv

Create a new virtual environment.
```
uv venv --python 3.11
```

Copy the following code into a file named hello_table_api.py.

# /// 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()

Run the following command to execute the Table API program from the directory where you created hello_table_api.py.
```
uv run hello_table_api.py
```

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

Prerequisites

Set up your environment and run a Table API program

Related content