Manage Flink Environments Using Confluent for Kubernetes

A Apache Flink® environment is a set of configurations that Flink applications and SQL statements share. After you create an environment, you can deploy Flink applications or run Flink SQL statements against it.

Create a Flink environment

A Flink environment is a set of configurations that Flink applications and SQL statements use.

The FlinkEnvironment specification supports the computePoolDefaults, kubernetesClusterName, secrets, and statementDefaults fields, and reports status information.

Create a FlinkEnvironment CR using the following specification, and deploy it with the kubectl apply -f command.

apiVersion: platform.confluent.io/v1beta1
kind: FlinkEnvironment
metadata:
  name:
  namespace:
spec:
  kubernetesNamespace:      --- [1]
  flinkApplicationDefaults: --- [2]
    metadata:               --- [3]
    spec:                   --- [4]
      flinkConfiguration:
  cmfRestClassRef:          --- [5]
    name:
    namespace:

[1] The namespace of the Flink cluster.
Typically, you would install the FlinkEnvironment CR in the CFK namespace (metadata.namespace), but the Flink cluster would be in another namespace (spec.kubernetesNamespace), for example, default.
[2] Configurations for the Flink cluster to specify the deployment-wide default application settings.
[3] Kubernetes API metadata.
[4] Spec of the FlinkApplicationSpec type.
[5] The reference to the REST Class you created in Create a CMF REST Class.
You can install FlinkEnvironment CR and the CMF REST class in different namespaces.
If omitted, the CMFRestClass of the name default in the same namespace is used.

An example FlinkEnvironment CR:

apiVersion: platform.confluent.io/v1beta1
kind: FlinkEnvironment
metadata:
  name: flink-env1
  namespace: operator
spec:
  kubernetesNamespace: default
  flinkApplicationDefaults:
    metadata:
      labels:
        "acmecorp.com/owned-by": "analytics-team"
    spec:
      flinkConfiguration:
        taskmanager.numberOfTaskSlots: "2"
        rest.profiling.enabled: "true"
  cmfRestClassRef:
    name: default
    namespace: operator

Check the status.

kubectl get flinkEnvironment -n <namespace> -oyaml