<a id="cmf-sql-examples-setup"></a>

# Set Up Your Environment for the Examples

The Flink SQL examples share one foundation: object storage, artifact
management and the environment catalog enabled, an environment, and a compute
pool. Complete this setup once, then run any example without repeating it.

Every command here is copy-paste runnable. These instructions assume CMF runs
in a namespace named `flink` and is reachable at `http://cmf:8080`, and they
use demo credentials (`console` / `console123`). Replace all three for any
shared or production use.

## Prerequisites

- A running Confluent Manager for Apache Flink (CMF) instance. To install one, see
  [Install Confluent Manager for Apache Flink with Helm](../../../installation/helm.md#install-cmf-helm).
- `kubectl` and `helm` access to the namespace where CMF runs.
- The `confluent` CLI, and `curl` for artifact uploads.
- JDK 11 or later and Maven 3.x, for the examples that build a JAR.

## Install object storage

Artifact management and the object-storage examples need an S3-compatible store.
Install MinIO with a `flink` bucket. Skip this step if you already have a store,
and adjust the endpoint and credentials in the later steps to match.

```bash
helm repo add minio https://charts.min.io/
helm install minio minio/minio \
  --namespace flink \
  --set rootUser=console --set rootPassword=console123 \
  --set 'buckets[0].name=flink' \
  --set mode=standalone --set replicas=1 \
  --set resources.requests.memory=500Mi \
  --set service.type=ClusterIP
```

MinIO is then reachable in-cluster at `http://minio.flink.svc.cluster.local:9000`.
To browse the bucket from your machine, port-forward the service and add an
`mc` alias (the `filesystem` and `iceberg` examples use it):

```bash
kubectl -n flink port-forward svc/minio 9000:9000 &
mc alias set minio http://localhost:9000 console console123
```

## Enable artifact management and the environment catalog

Every example uploads an artifact, so enable artifact management and point it at
the object store. Enable the environment catalog and the
`examples` catalog that the [Redact a Column with a User-Defined Function](udf.md#cmf-sql-example-udf) example queries. Both
catalogs are off by default. Save this as `examples-values.yaml`:

```yaml
cmf:
  artifacts:
    enabled: true
    basePath: s3://flink/cmf-artifacts
    configuration:
      s3.endpoint: http://minio.flink.svc.cluster.local:9000
      s3.path.style.access: "true"
      s3.connection.ssl.enabled: "false"
      s3.access-key: console
      s3.secret-key: console123
  sql:
    environmentCatalog:
      enabled: true
    examplesCatalog:
      enabled: true
```

Upgrade your CMF release with these values. `cmf` is the release name here.
Without `--reuse-values`, `helm upgrade` resets every value you do not pass,
such as your authentication and encryption settings:

```bash
helm upgrade --install cmf confluentinc/confluent-manager-for-apache-flink --version "~2.4.0" \
  --namespace flink \
  --reuse-values \
  --values examples-values.yaml
```

## Create an environment

Create an environment named `test`. CMF exposes its environment catalog as
`_env_test` with a `default` database, which the examples set as the current
catalog and database:

```bash
confluent flink environment create test \
  --kubernetes-namespace flink \
  --url http://cmf:8080
```

## Create a compute pool

Create a general-purpose compute pool named `pool` on the `cp-flink-sql` image.
The pool connects to the object store you set up above for checkpoints. CMF
requires checkpointing to be enabled for `INSERT INTO` statements, and it also
gives streaming jobs durable state. The [Read CSV Files from Object Storage](filesystem.md#cmf-sql-example-filesystem) and
[Round-Trip an Apache Iceberg Table](iceberg.md#cmf-sql-example-iceberg) examples create their own specialized pools.
Every other example runs on this one. Save this as `pool.json`:

```json
{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "ComputePool",
  "metadata": { "name": "pool" },
  "spec": {
    "type": "DEDICATED",
    "clusterSpec": {
      "flinkVersion": "v1_19",
      "image": "confluentinc/cp-flink-sql:1.19-cp11",
      "flinkConfiguration": {
        "taskmanager.numberOfTaskSlots": "4",
        "execution.checkpointing.interval": "10000",
        "state.checkpoints.dir": "s3://flink/pool/checkpoints",
        "state.savepoints.dir": "s3://flink/pool/savepoints",
        "s3.endpoint": "http://minio.flink.svc.cluster.local:9000",
        "s3.path.style.access": "true",
        "s3.connection.ssl.enabled": "false",
        "s3.access-key": "console",
        "s3.secret-key": "console123"
      },
      "taskManager": { "resource": { "cpu": 0.5, "memory": "1024m" } },
      "jobManager": { "resource": { "cpu": "500m", "memory": "1536m" } }
    }
  }
}
```

```bash
confluent flink compute-pool create pool.json \
  --environment test \
  --url http://cmf:8080
```

Your environment is ready. Continue to any example in
[Flink SQL Examples in Confluent Manager for Apache Flink](index.md#cmf-sql-examples).

## Related content

- [Install Confluent Manager for Apache Flink with Helm](../../../installation/helm.md#install-cmf-helm)
- [Manage Artifacts in Confluent Manager for Apache Flink](../../../configure/artifacts.md#cmf-artifacts)
- [Manage Environment Catalogs in Confluent Manager for Apache Flink](../../../configure/environment-catalog.md#cmf-environment-catalog)
- [Flink SQL Examples in Confluent Manager for Apache Flink](index.md#cmf-sql-examples)