Manage Flink SQL Statements Using Confluent for Kubernetes

Beyond Apache Flink® applications, you can use Confluent for Kubernetes (CFK) to run Flink SQL statements through Confluent Manager for Apache Flink® (CMF). This topic explains the Day-2 custom resources that make up a SQL deployment, the order in which you apply them, and an end-to-end walkthrough that builds a streaming SQL job from a fresh CFK install.

Flink SQL introduces a set of custom resources (CRs) that map to the objects in the Flink SQL model:

FlinkSecret syncs a Kubernetes Secret that holds Apache Kafka® or Schema Registry credentials to CMF.
FlinkEnvironmentSecretMapping exposes a synced secret to a Flink environment so that catalogs and databases can reference it.
FlinkKafkaCatalog binds a Schema Registry instance for the databases in the catalog.
FlinkKafkaDatabase is a SQL database inside a catalog, mapped to a Kafka cluster.
FlinkComputePool is the compute that runs statements. A pool is either DEDICATED or SHARED.
FlinkStatement is a single Flink SQL statement, which runs as a streaming job.

Note

Flink SQL support in CFK is a preview feature in CFK 3.3.0. Do not use preview features in production.

Requirements and considerations

To run Flink SQL statements with CFK, you need the following:

CFK 3.3.0 or later, installed with both Day-2 feature flags enabled:
- enableCMFDay2Ops=true turns on the CMFRestClass, FlinkEnvironment, and FlinkApplication resources.
- enableFlinkSQL=true adds the SQL resources described in this topic (FlinkSecret, FlinkEnvironmentSecretMapping, FlinkKafkaCatalog, FlinkKafkaDatabase, FlinkComputePool, and FlinkStatement).
CMF 2.3.0 or later, exposing the Flink SQL REST API.
The Flink SQL runtime image that matches your CMF version, set on each compute pool through clusterSpec.image. This example uses confluentinc/cp-flink-sql:1.19-cp8. If the image does not match your CMF version, the JobManager fails to load the statement plan.
CMF installed with secret encryption enabled (--set encryption.enabled=true). A FlinkSecret syncs credentials into CMF, so CMF must be able to store them encrypted at rest before you roll out any SQL Day-2 resources.
A Kafka cluster and a Schema Registry instance for the catalog and database to connect to.

Flink SQL resource chain

The SQL resources form a dependency chain. Apply them in the following order, and delete them in the reverse order so that each CMF-side resource is removed before the resource it depends on:

FlinkSecret
  -> FlinkEnvironmentSecretMapping
    -> FlinkKafkaCatalog
      -> FlinkKafkaDatabase
        -> FlinkComputePool
          -> FlinkStatement

FlinkSecret through FlinkKafkaDatabase are shared across environments and statements. Create them as described in Manage Global Flink Resources Using Confluent for Kubernetes. This topic covers the FlinkComputePool and FlinkStatement resources that run on top of them.

CFK enforces the following cross-resource validation rules through Common Expression Language (CEL) rules on the CRDs:

FlinkComputePool.spec.type is immutable after creation.
FlinkComputePool.spec.state is valid only on SHARED pools. A CEL rule on the CRD rejects state on a DEDICATED pool.
FlinkStatement.spec.statement is immutable after the statement is running. To change the SQL, create a new FlinkStatement.

In addition, CMF applies the following runtime behavior:

Data Definition Language (DDL) statements, such as CREATE TABLE, run only in the environments listed in the database’s ddlEnvironments.

When to use FlinkStatement compared to FlinkApplication

Both FlinkStatement and FlinkApplication run Flink workloads through CMF, but they target different use cases:

Use a FlinkStatement to run Flink SQL. Each statement is a single SQL query that CMF compiles and runs as a Flink job. This is the right choice for SQL-based stream processing, such as joins, aggregations, and materialized views, where you do not manage application code or JARs.
Use a FlinkApplication to run a packaged Flink program. An application points to a JAR or container image that contains compiled Flink code. This is the right choice for the DataStream or Table API, or any logic that cannot be expressed as SQL.

To manage Flink applications, see Manage Flink Applications Using Confluent for Kubernetes.

Set up prerequisites

Before you create the SQL resources, install the operators and the platform resources that the SQL chain depends on.

Install the Flink Kubernetes operator and CMF. For instructions, see Install Confluent Manager for Apache Flink. When you deploy CMF, enable secret encryption:

helm upgrade --install cmf \
  confluentinc/confluent-manager-for-apache-flink \
  --version 2.3.0 \
  --set encryption.enabled=true \
  --namespace operator

Install or upgrade CFK with both Day-2 flags enabled:

helm upgrade --install confluent-operator \
  confluentinc/confluent-for-kubernetes \
  --set enableCMFDay2Ops=true \
  --set enableFlinkSQL=true

Create a CMFRestClass. The SQL resources communicate with CMF through this REST class. For the steps, see Create a CMF REST Class.
Create a FlinkEnvironment. Every SQL resource except FlinkSecret references this environment by name through spec.flinkEnvironment. For the steps, see Create a Flink environment.
Deploy a Kafka cluster and a Schema Registry instance for the catalog and database to connect to.
Create the global Flink resources that the SQL chain depends on the FlinkSecret, FlinkEnvironmentSecretMapping, FlinkKafkaCatalog, and FlinkKafkaDatabase. For the steps, see Manage Global Flink Resources Using Confluent for Kubernetes.

Create a FlinkComputePool

A statement runs on a compute pool. A DEDICATED pool owns a fixed set of resources reserved for its statements. A SHARED pool multiplexes statements across a common, elastically managed set of resources and accepts spec.state to pause or resume scheduling.

Create one or both compute pools, and deploy them with the kubectl apply -f command.

A DEDICATED pool:

apiVersion: platform.confluent.io/v1beta1
kind: FlinkComputePool
metadata:
  name: dedicated-pool
  namespace: operator
spec:
  type: DEDICATED                           --- [1]
  flinkEnvironment: flink-env1
  clusterSpec:                              --- [2]
    flinkVersion: v1_19
    image: confluentinc/cp-flink-sql:1.19-cp8
    jobManager:
      resource:
        memory: "1024m"
        cpu: 0.5
    taskManager:
      resource:
        memory: "1024m"
        cpu: 0.5
  cmfRestClassRef:
    name: default
    namespace: operator

A SHARED pool:

apiVersion: platform.confluent.io/v1beta1
kind: FlinkComputePool
metadata:
  name: shared-pool
  namespace: operator
spec:
  type: SHARED
  state: RUNNING                            --- [3]
  flinkEnvironment: flink-env1
  clusterSpec:
    flinkVersion: v1_19
    image: confluentinc/cp-flink-sql:1.19-cp8
    jobManager:
      resource:
        memory: "1024m"
        cpu: 0.5
    taskManager:
      resource:
        memory: "1024m"
        cpu: 0.5
  cmfRestClassRef:
    name: default
    namespace: operator

[1] The pool type, either DEDICATED or SHARED. This field is immutable after creation.
[2] The Flink version, the statement runtime image, and the JobManager and TaskManager resources. Specify the JobManager and TaskManager resources, including jobManager.resource.memory. Otherwise, the deployment can fail when Flink starts the pool.
[3] Valid only on SHARED pools. Set state to SUSPENDED to pause scheduling without deleting the pool.

Check the status:

kubectl get flinkcomputepool -n operator

Create the source and sink tables

Before you run a streaming statement, create the tables that it reads from and writes to. A CREATE TABLE statement runs as a FlinkStatement and is gated by the database’s ddlEnvironments.

On a Kafka-backed catalog, the connector is confluent, and the bootstrap servers and Schema Registry come from the catalog binding. Each table maps to a Kafka topic with its schema in Schema Registry. Because CREATE TABLE is DDL, each statement runs once and reaches status.phase: COMPLETED.

Create the source and sink tables, and deploy them with the kubectl apply -f command.

apiVersion: platform.confluent.io/v1beta1
kind: FlinkStatement
metadata:
  name: create-pageviews
  namespace: operator
spec:
  flinkEnvironment: flink-env1
  computePoolName: dedicated-pool
  statement: |
    CREATE TABLE `kafka-catalog`.`clickstream`.`pageviews` (
      `user_id` INT,
      `page_id` STRING
    ) WITH (
      'connector' = 'confluent'
    );
  cmfRestClassRef:
    name: default
    namespace: operator
---
apiVersion: platform.confluent.io/v1beta1
kind: FlinkStatement
metadata:
  name: create-pageviews-by-user
  namespace: operator
spec:
  flinkEnvironment: flink-env1
  computePoolName: dedicated-pool
  statement: |
    CREATE TABLE `kafka-catalog`.`clickstream`.`pageviews_by_user` (
      `user_id` INT NOT NULL,
      `view_count` BIGINT,
      PRIMARY KEY (`user_id`) NOT ENFORCED
    ) WITH (
      'connector' = 'confluent',
      'changelog.mode' = 'upsert'
    );
  cmfRestClassRef:
    name: default
    namespace: operator

Check the status:

kubectl get flinkstatement create-pageviews create-pageviews-by-user -n operator

Create a FlinkStatement

A FlinkStatement runs a single Flink SQL statement on a compute pool within an environment. The following streaming INSERT continuously aggregates the pageviews source into the pageviews_by_user sink.

Create the FlinkStatement CR and deploy it with the kubectl apply -f command.

apiVersion: platform.confluent.io/v1beta1
kind: FlinkStatement
metadata:
  name: pageviews-by-user
  namespace: operator
spec:
  flinkEnvironment: flink-env1
  computePoolName: dedicated-pool          --- [1]
  flinkConfiguration:                       --- [2]
    execution.checkpointing.interval: "10s"
  statement: |                              --- [3]
    INSERT INTO `kafka-catalog`.`clickstream`.`pageviews_by_user`
    SELECT user_id, COUNT(*) AS view_count
    FROM `kafka-catalog`.`clickstream`.`pageviews`
    GROUP BY user_id;
  cmfRestClassRef:
    name: default
    namespace: operator

[1] The compute pool that runs the statement.
[2] A streaming INSERT requires checkpointing. Set the checkpointing interval here or on the compute pool.
[3] The SQL statement. This field is immutable after the statement is running. Set spec.stopped: true to stop the statement without deleting it.

Check the status:
```
kubectl get flinkstatement pageviews-by-user -n operator -oyaml
```
Expect cfkInternalState: CREATED and cmfSync.status: Created. CMF runs the statement as a Flink job in the environment’s Kubernetes namespace (default in this example). Confirm that the job is running by checking its FlinkDeployment:
```
kubectl get flinkdeployment pageviews-by-user -n default \
  -o jsonpath='{.status.jobStatus.state}{"\n"}'
```
The pageviews topic starts empty, so the job runs but emits nothing until rows arrive. Seed rows with an INSERT INTO ... pageviews VALUES (...) statement, or point the source at an existing topic.

Delete the SQL resources

Delete the chain in the reverse order of creation so that each CMF-side resource is removed before the resource it depends on. Wait for each statement and pool to drain its Flink job before you uninstall CMF and the Flink operator. The Flink operator owns the FlinkDeployment finalizer, so removing it before the jobs drain orphans the jobs.

kubectl delete flinkstatement pageviews-by-user -n operator
kubectl delete flinkstatement create-pageviews create-pageviews-by-user -n operator
kubectl delete flinkcomputepool shared-pool dedicated-pool -n operator
kubectl delete flinkkafkadatabase clickstream -n operator
kubectl delete flinkkafkacatalog kafka-catalog -n operator
kubectl delete flinkenvironmentsecretmapping flink-connection-secret -n operator
kubectl delete flinksecret flink-connection-secret -n operator