<a id="cpf-get-started"></a>

# Get Started with Applications in Confluent Manager for Apache Flink

Confluent Platform for Apache Flink® is compatible with open-source Apache Flink®. To submit an application you must use the
Helm chart to install the Confluent Manager for Apache Flink (CMF) application and the Flink Kubernetes operator.
After you have installed both, you can deploy a Flink job.
The next few sections will describe necessary prerequisites and walk you through the
steps to install CMF and deploy a Flink job.

For requirement and compatibility details, see [Versions and Interoperability for Confluent Manager for Apache Flink](../installation/versions-interoperability.md#cmf-interop).

## Prerequisites

To use Confluent Platform for Apache Flink, you must meet the following prerequisites:

### Kubernetes

Confluent Platform for Apache Flink® is designed for Kubernetes. Therefore, you must have access to a Kubernetes cluster to deploy the Flink Kubernetes operator and individual Flink jobs:

- A Kubernetes cluster running a supported version. For a list of supported versions, see [Versions and Interoperability for Confluent Manager for Apache Flink](../installation/versions-interoperability.md#cmf-interop).
- [kubectl](https://kubernetes.io/docs/tasks/tools/install-kubectl/) installed, initialized,
  with the context set. You also must have the `kubeconfig` file configured for your cluster.
- [Helm 3](https://helm.sh/docs/intro/install/) installed.

### Confluent CLI

If you have not installed Confluent Platform 2.4 you must install the latest version of the Confluent CLI.
In either case, make sure to set the `CONFLUENT_HOME`
environment variable and add the bin directory to your path.
For more information, see [Install the Confluent CLI](https://docs.confluent.io/confluent-cli/current/install.html).

Set the `CONFLUENT_HOME` environment variable.

```none
export CONFLUENT_HOME=<The directory where Confluent is installed>
```

Add the location of the CLI (the `/bin` directory) to the PATH:

```none
export PATH=$CONFLUENT_HOME/bin:$PATH
```

### Durable storage

In production, Flink requires durable storage that is readable and writeable from each Flink cluster.
Durable storage is used to store consistent snapshots of the state of your Flink jobs (checkpoints and savepoints).

Examples of durable storage include:

- S3-compatible storage (S3, MinIO)
- Microsoft Azure Blob Storage
- Google Cloud Storage

For more information on configuring durable storage, see [How to Set Up Checkpointing to AWS S3 with Confluent Manager for Apache Flink](../how-to-guides/checkpoint-s3.md#cmf-s3-checkpointing).

### Supported components

Check to make sure that you are using components supported by Confluent Platform for Apache Flink.
See, [Confluent Platform for Apache Flink Features and Support](../jobs/applications/supported-features.md#cpflink-vs-oss).

<a id="step-install-cmf"></a>

## Step 1: Install Confluent Manager for Apache Flink

To install Confluent Manager for Apache Flink, you have to use Helm to add the Confluent repository, install the certificate
manager and then install the Flink Kubernetes operator, and finally install CMF.

#### WARNING
This quickstart guide assumes that everything will be installed into the same Kubernetes
namespace. In this example it is `default`. You can use a different namespace, as long as it is
consistent across all the commands. Please refer to the [Helm installation guide](../installation/helm.md#install-cmf-helm)
for more information on how to setup Confluent Platform for Apache Flink in a multi-namespace environment, in a production
environment or on OpenShift.

1. Add the Confluent Platform repository.
   ```none
   helm repo add confluentinc https://packages.confluent.io/helm
   helm repo update
   ```
2. Install the certificate manager.
   ```none
   kubectl create -f https://github.com/jetstack/cert-manager/releases/download/v1.18.2/cert-manager.yaml
   ```
3. Install the Flink Kubernetes Operator
   ```none
   helm upgrade --install --version "~1.150.0" cp-flink-kubernetes-operator confluentinc/flink-kubernetes-operator
   ```
4. Install Confluent Manager for Apache Flink.
   ```none
   helm upgrade --install cmf --version "~2.4.0" \
   confluentinc/confluent-manager-for-apache-flink \
   --namespace default --set encryption.enabled=false
   ```

#### NOTE
The `encryption.enabled=false` setting initializes the CMF database without encryption. You
cannot enable encryption later on. A new CMF installation with a new database is required to
set up encryption.

1. Check that everything deployed correctly.
   ```none
   kubectl get pods
   ```

<a id="step-deploy-jobs"></a>

## Step 2: Deploy Flink jobs

To deploy Flink jobs, you should open port forwarding to CMF, and create an environment and application.
You can then use the Web UI to confirm that your application has been created.

1. Open port forwarding to CMF.
   ```bash
   kubectl port-forward svc/cmf-service 8080:80
   ```
2. Create an environment using the Confluent CLI. For a list of Confluent CLI commands
   for Confluent Platform for Apache Flink, see [confluent flink (On-Premises tab)](https://docs.confluent.io/confluent-cli/current/command-reference/flink/).
   ```bash
   confluent flink environment create env1 --url http://localhost:8080 --kubernetes-namespace default
   ```
3. Create the JSON application file. Following is an example.
   ```json
   {
   "apiVersion": "cmf.confluent.io/v1",
   "kind": "FlinkApplication",
   "metadata": {
      "name": "basic-example"
   },
   "spec": {
      "flinkConfiguration": {
         "metrics.reporter.prom.factory.class": "org.apache.flink.metrics.prometheus.PrometheusReporterFactory",
         "metrics.reporter.prom.port": "9249-9250",
         "taskmanager.numberOfTaskSlots": "4"
      },
      "flinkVersion": "v2_0",
      "image": "confluentinc/cp-flink:2.0.1-cp1",
      "job": {
         "jarURI": "local:///opt/flink/examples/streaming/StateMachineExample.jar",
         "parallelism": 2,
         "state": "running",
         "upgradeMode": "stateless"
      },
      "jobManager": {
         "resource": {
         "cpu": 1,
         "memory": "1024m"
         }
      },
      "serviceAccount": "flink",
      "taskManager": {
         "resource": {
         "cpu": 1,
         "memory": "1024m"
         }
      }
   }
   }
   ```
4. Use the application command, passing the `application.json file` to create the application.
   ```bash
   confluent flink application create application.json --environment env1 --url http://localhost:8080
   ```
5. Access the Flink Web UI to confirm you have successfully created the application:
   ```bash
   confluent flink application web-ui-forward basic-example --environment env1 --port 8090 --url http://localhost:8080
   ```

## Step 3: Cleanup

<a id="cpf-get-started-cleanup"></a>
1. Delete the Flink Application:
   ```bash
   confluent flink application delete basic-example --environment env1 --url http://localhost:8080
   ```
2. Uninstall the Flink Kubernetes Operator and CMF:
   ```bash
   helm uninstall cp-flink-kubernetes-operator
   helm uninstall cmf
   ```

## Related content

- [Confluent Platform for Apache Flink Features and Support](../jobs/applications/supported-features.md#cpflink-vs-oss)
- [Get Help with Confluent Platform for Apache Flink](../get-help.md#cpf-get-help)
- [Manage Flink With Confluent for Kubernetes](../clients-api/flink-cfk.md#cmf-cfk)
- [How to Secure a Flink Job with Confluent Manager for Apache Flink](../jobs/configure/security.md#cpf-security)