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.

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

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

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 in the same step. Save this as examples-values.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

Upgrade your CMF release with these values (cmf is the release name here):

helm upgrade --install cmf confluentinc/confluent-manager-for-apache-flink \
  --namespace flink \
  --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:

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 and Round-Trip an Apache Iceberg Table examples create their own specialized pools. Every other example runs on this one. Save this as pool.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" } }
    }
  }
}

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.

Related content