<a id="co-manage-flink-sql"></a>

<a id="co-flink-sql"></a>

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

<a id="co-flink-sql-requirements"></a>

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

<a id="co-flink-sql-chain"></a>

## 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:

```text
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](co-flink-global-resources.md#co-flink-global-resources). 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`.

<a id="co-flink-sql-vs-application"></a>

## 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](co-manage-flink.md#co-manage-flink).

<a id="co-flink-sql-prerequisites"></a>

## Set up prerequisites

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

1. Install the Flink Kubernetes operator and CMF. For instructions, see
   [Install Confluent Manager for Apache Flink](https://docs.confluent.io/platform/current/flink/installation/helm.html).
   When you deploy CMF, enable secret encryption:
   ```bash
   helm upgrade --install cmf \
     confluentinc/confluent-manager-for-apache-flink \
     --version 2.3.0 \
     --set encryption.enabled=true \
     --namespace operator
   ```
2. Install or upgrade CFK with both Day-2 flags enabled:
   ```bash
   helm upgrade --install confluent-operator \
     confluentinc/confluent-for-kubernetes \
     --set enableCMFDay2Ops=true \
     --set enableFlinkSQL=true
   ```
3. Create a CMFRestClass. The SQL resources communicate with CMF through this
   REST class. For the steps, see [Create a CMF REST Class](co-flink-overview.md#co-flink-rest-class).
4. Create a FlinkEnvironment. Every SQL resource except FlinkSecret references
   this environment by name through `spec.flinkEnvironment`. For the steps,
   see [Create a Flink environment](co-manage-flink-environments.md#co-flink-environment).
5. Deploy a Kafka cluster and a Schema Registry instance for the catalog and
   database to connect to.
6. 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](co-flink-global-resources.md#co-flink-global-resources).

<a id="co-flink-sql-compute-pool"></a>

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

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

   A `DEDICATED` pool:
   ```yaml
   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:
   ```yaml
   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.
2. Check the status:
   ```bash
   kubectl get flinkcomputepool -n operator
   ```

<a id="co-flink-sql-create-tables"></a>

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

1. Create the source and sink tables, and deploy them with the
   `kubectl apply -f` command.
   ```yaml
   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
   ```
2. Check the status:
   ```bash
   kubectl get flinkstatement create-pageviews create-pageviews-by-user -n operator
   ```

<a id="co-flink-sql-statement"></a>

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

1. Create the FlinkStatement CR and deploy it with the `kubectl apply -f`
   command.
   ```yaml
   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.
2. Check the status:
   ```bash
   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:
   ```bash
   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.

<a id="co-flink-sql-teardown"></a>

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

```bash
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
```