Round-Trip an Apache Iceberg Table

This example reads and writes an Apache Iceberg™ table on Confluent Manager for Apache Flink (CMF) from end to end. You create an object storage, configure CMF and build the Iceberg connector, then create a table, write three rows, and read them back.

Important

Reading an Iceberg table back through Flink SQL requires CMF 2.4.2 or later. Writing requires a cp-flink-sql image that carries the connector classloader fix (1.19-cp10 or later). Use both, or the SELECT step fails to load the Iceberg catalog.

Prerequisites

Complete the examples setup first. It installs the MinIO object store and enables artifact management, which this example builds on. Iceberg needs extra CMF configuration and a dedicated compute pool, both covered below. The credentials (console / console123) and namespace (flink) match the setup. Substitute your own.

Why Iceberg needs configuration on both sides

Iceberg’s Hadoop catalog runs in two places, so both need to reach the object store. It runs inside the Flink cluster (the INSERT and SELECT jobs), and it runs inside CMF itself when CMF compiles a statement (CREATE TABLE writes the table’s first metadata file, and every statement loads the catalog to plan the job). You give the Flink side its credentials through the compute pool, and the CMF side through the configuration in the next section.

Configure CMF for Iceberg

CMF needs S3 access on its compile path, a larger artifact upload limit, and a wider class allowlist. First create a core-site.xml ConfigMap in CMF’s namespace so CMF’s Iceberg catalog can reach the object store (CMF is not a Flink process, so it reads Hadoop settings from HADOOP_CONF_DIR rather than from a pool’s flink.hadoop.* keys):

apiVersion: v1
kind: ConfigMap
metadata:
  name: cmf-hadoop-conf
  namespace: flink
data:
  core-site.xml: |
    <?xml version="1.0"?>
    <configuration>
      <property><name>fs.s3a.endpoint</name><value>http://minio.flink.svc.cluster.local:9000</value></property>
      <property><name>fs.s3a.path.style.access</name><value>true</value></property>
      <property><name>fs.s3a.connection.ssl.enabled</name><value>false</value></property>
      <property><name>fs.s3a.access.key</name><value>console</value></property>
      <property><name>fs.s3a.secret.key</name><value>console123</value></property>
      <property><name>fs.s3a.aws.credentials.provider</name><value>org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider</value></property>
      <property><name>fs.s3a.impl</name><value>org.apache.hadoop.fs.s3a.S3AFileSystem</value></property>
    </configuration>

Apply it, then upgrade CMF with the following iceberg-values.yaml. These values extend the examples setup values with the Iceberg-specific settings, so this single upgrade is all the Iceberg example needs: it raises maxUploadSize above the ~112 MB connector JAR, adds the JDK packages the Iceberg connector loads at compile time to the class allowlist, and mounts the ConfigMap with the AWS environment the S3A credential chain reads.

cmf:
  artifacts:
    enabled: true
    basePath: s3://flink/cmf-artifacts
    maxUploadSize: 300MB
    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
    # The default allowlist plus the JDK packages a Hadoop/Iceberg connector
    # loads at compile time (org.w3c.dom, org.xml.sax, org.ietf.jgss, sun.misc).
    # Helm replaces this list, so it must repeat the defaults.
    classAllowlist:
      - 'java\..*'
      - 'javax\..*'
      - 'jdk\.internal\..*'
      - 'org\.apache\.flink\..*'
      - 'io\.confluent\.flink\..*'
      - 'org\.slf4j\..*'
      - 'org\.apache\.kafka\..*'
      - 'org\.w3c\.dom\..*'
      - 'org\.xml\.sax\..*'
      - 'org\.ietf\.jgss\..*'
      - 'sun\.misc\..*'
extraEnv:
  - name: HADOOP_CONF_DIR
    value: /opt/hadoop-conf
  - name: AWS_ACCESS_KEY_ID
    value: console
  - name: AWS_SECRET_ACCESS_KEY
    value: console123
  - name: AWS_REGION
    value: us-east-1
mountedVolumes:
  volumes:
    - name: hadoop-conf
      configMap:
        name: cmf-hadoop-conf
  volumeMounts:
    - name: hadoop-conf
      mountPath: /opt/hadoop-conf

kubectl apply -f cmf-hadoop-conf.yaml

helm upgrade --install cmf confluentinc/confluent-manager-for-apache-flink \
  --namespace flink \
  --values iceberg-values.yaml

Build the connector JAR

The connector is a fat JAR that bundles the Iceberg Flink runtime, an S3A filesystem, and a slim AWS SDK. Iceberg’s published iceberg-flink-runtime JAR carries neither hadoop-aws nor an AWS SDK, so an S3 warehouse needs those shaded in. The bundle also relocates Flink’s HadoopUtils class so the connector resolves Hadoop from its own JAR rather than CMF’s Hadoop-free host classpath.

The versions are load-bearing: iceberg-flink-runtime-1.19 1.10.2 is the last Iceberg release that supports Flink 1.19, and the AWS SDK version must match the one hadoop-aws was built against. All dependencies are on Maven Central. pom.xml:

<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>

  <groupId>io.confluent.examples.flink</groupId>
  <artifactId>iceberg-flink-1.19-s3-bundle</artifactId>
  <version>1.0.0</version>
  <packaging>jar</packaging>

  <properties>
    <maven.compiler.release>11</maven.compiler.release>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <!-- 1.10.2 is the last Iceberg release that supports Flink 1.19. -->
    <iceberg.version>1.10.2</iceberg.version>
    <hadoop.version>3.4.1</hadoop.version>
    <flink.version>1.19.1</flink.version>
    <!-- The AWS SDK v2 version hadoop-aws 3.4.1 was built against; pin it exactly. -->
    <awssdk.version>2.24.6</awssdk.version>
  </properties>

  <dependencyManagement>
    <dependencies>
      <dependency>
        <groupId>software.amazon.awssdk</groupId>
        <artifactId>bom</artifactId>
        <version>${awssdk.version}</version>
        <type>pom</type>
        <scope>import</scope>
      </dependency>
    </dependencies>
  </dependencyManagement>

  <dependencies>
    <!-- The 'iceberg' factory (FlinkDynamicTableFactory) and the HadoopCatalog. -->
    <dependency>
      <groupId>org.apache.iceberg</groupId>
      <artifactId>iceberg-flink-runtime-1.19</artifactId>
      <version>${iceberg.version}</version>
    </dependency>

    <!-- Hadoop client (Configuration, FileSystem); needed at compile and runtime. -->
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-client-api</artifactId>
      <version>${hadoop.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-client-runtime</artifactId>
      <version>${hadoop.version}</version>
    </dependency>

    <!-- S3AFileSystem (s3a://). Exclude hadoop-common/-client (provided by
         hadoop-client-runtime) and the large AWS SDK bundle; add slim modules below. -->
    <dependency>
      <groupId>org.apache.hadoop</groupId>
      <artifactId>hadoop-aws</artifactId>
      <version>${hadoop.version}</version>
      <exclusions>
        <exclusion>
          <groupId>org.apache.hadoop</groupId>
          <artifactId>hadoop-common</artifactId>
        </exclusion>
        <exclusion>
          <groupId>org.apache.hadoop</groupId>
          <artifactId>hadoop-client</artifactId>
        </exclusion>
        <exclusion>
          <groupId>software.amazon.awssdk</groupId>
          <artifactId>bundle</artifactId>
        </exclusion>
      </exclusions>
    </dependency>

    <!-- Slim AWS SDK v2 modules S3A needs (vs. the full bundle); versions from the BOM. -->
    <dependency>
      <groupId>software.amazon.awssdk</groupId>
      <artifactId>s3</artifactId>
    </dependency>
    <dependency>
      <groupId>software.amazon.awssdk</groupId>
      <artifactId>s3-transfer-manager</artifactId>
    </dependency>
    <dependency>
      <groupId>software.amazon.awssdk</groupId>
      <artifactId>apache-client</artifactId>
    </dependency>
    <dependency>
      <groupId>software.amazon.awssdk</groupId>
      <artifactId>sts</artifactId>
    </dependency>

    <!-- Source of org.apache.flink.runtime.util.HadoopUtils; only that class is
         kept and it is relocated (see below). -->
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-hadoop-fs</artifactId>
      <version>${flink.version}</version>
    </dependency>
  </dependencies>

  <build>
    <finalName>iceberg-flink-1.19-s3-bundle-${project.version}</finalName>
    <plugins>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-shade-plugin</artifactId>
        <version>3.5.3</version>
        <executions>
          <execution>
            <phase>package</phase>
            <goals><goal>shade</goal></goals>
            <configuration>
              <createDependencyReducedPom>false</createDependencyReducedPom>
              <filters>
                <!-- Keep only HadoopUtils from flink-hadoop-fs. -->
                <filter>
                  <artifact>org.apache.flink:flink-hadoop-fs</artifact>
                  <includes>
                    <include>org/apache/flink/runtime/util/HadoopUtils*.class</include>
                  </includes>
                </filter>
                <filter>
                  <artifact>*:*</artifact>
                  <excludes>
                    <exclude>META-INF/*.SF</exclude>
                    <exclude>META-INF/*.DSA</exclude>
                    <exclude>META-INF/*.RSA</exclude>
                    <exclude>module-info.class</exclude>
                    <exclude>META-INF/versions/*/module-info.class</exclude>
                  </excludes>
                </filter>
              </filters>
              <transformers>
                <!-- Merge the Flink Factory SPI (iceberg) and the Hadoop FileSystem SPI (s3a). -->
                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
              </transformers>
              <!-- Relocate HadoopUtils so it no longer matches org.apache.flink.* and
                   loads from this JAR instead of CMF's Hadoop-free host classpath. -->
              <relocations>
                <relocation>
                  <pattern>org.apache.flink.runtime.util.HadoopUtils</pattern>
                  <shadedPattern>shaded.flink.runtime.util.HadoopUtils</shadedPattern>
                </relocation>
              </relocations>
            </configuration>
          </execution>
        </executions>
      </plugin>
    </plugins>
  </build>
</project>

Build it:

mvn clean package

The build writes target/iceberg-flink-1.19-s3-bundle-1.0.0.jar (about 112 MB).

Create the compute pool

Create a compute pool on the cp-flink-sql:1.19-cp11 image whose configuration carries the Hadoop S3A credentials the Iceberg catalog uses on the Flink side, not only the Flink s3.* keys. Save this as iceberg-pool.json:

{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "ComputePool",
  "metadata": { "name": "iceberg-pool" },
  "spec": {
    "type": "DEDICATED",
    "clusterSpec": {
      "flinkVersion": "v1_19",
      "image": "confluentinc/cp-flink-sql:1.19-cp11",
      "flinkConfiguration": {
        "taskmanager.numberOfTaskSlots": "4",
        "execution.checkpointing.interval": "2000",
        "state.checkpoints.dir": "s3://flink/iceberg-pool/checkpoints",
        "state.savepoints.dir": "s3://flink/iceberg-pool/savepoints",
        "high-availability.storageDir": "s3://flink/iceberg-pool/recovery",
        "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",
        "flink.hadoop.fs.s3a.endpoint": "http://minio.flink.svc.cluster.local:9000",
        "flink.hadoop.fs.s3a.path.style.access": "true",
        "flink.hadoop.fs.s3a.connection.ssl.enabled": "false",
        "flink.hadoop.fs.s3a.access.key": "console",
        "flink.hadoop.fs.s3a.secret.key": "console123",
        "flink.hadoop.fs.s3a.aws.credentials.provider": "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider"
      },
      "taskManager": { "resource": { "cpu": 0.5, "memory": "1024m" } },
      "jobManager": { "resource": { "cpu": "500m", "memory": "1536m" } }
    }
  }
}

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

Upload the connector artifact

Label the JAR as a connector with the iceberg identifier, in artifact.json:

{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "Artifact",
  "metadata": {
    "name": "iceberg-connector.jar",
    "labels": {
      "cmf.confluent.io/sql-artifact-type": "connector",
      "cmf.confluent.io/sql-artifact-identifier": "iceberg"
    }
  },
  "spec": {}
}

Upload it. The JAR is large, so allow for the upload time:

curl -X POST http://cmf:8080/cmf/api/v1/environments/test/artifacts \
  -F 'artifact=@artifact.json;type=application/json' \
  -F 'file=@target/iceberg-flink-1.19-s3-bundle-1.0.0.jar'

The response reaches status.phase of READY.

Create the table

Create an Iceberg table in the environment catalog with a Hadoop catalog on the object store. CREATE TABLE stores the options only; the connector is not resolved yet.

confluent flink statement create ice-create \
  --url http://cmf:8080 \
  --environment test \
  --compute-pool iceberg-pool \
  --catalog _env_test \
  --database default \
  --sql "CREATE TABLE \`ice_orders\` (\`id\` BIGINT, \`data\` STRING) WITH ('connector' = 'iceberg', 'catalog-type' = 'hadoop', 'catalog-name' = 'hadoop_s3', 'warehouse' = 's3a://flink/iceberg-warehouse');"

The statement reaches COMPLETED with sqlKind of CREATE_TABLE.

Write rows

Insert three rows. This is the first statement to use the table, so CMF resolves the iceberg artifact and ships it to the Flink cluster. The bounded INSERT reaches COMPLETED when the rows are written.

confluent flink statement create ice-insert \
  --url http://cmf:8080 \
  --environment test \
  --compute-pool iceberg-pool \
  --catalog _env_test \
  --database default \
  --sql "INSERT INTO \`ice_orders\` VALUES (1, 'a'), (2, 'b'), (3, 'c');"

Confirm the data landed in object storage. The Hadoop catalog lays the table out under <database>/<table>/:

kubectl -n flink port-forward svc/minio 9000:9000 &
mc alias set minio http://localhost:9000 console console123
mc ls --recursive minio/flink/iceberg-warehouse/

default/ice_orders/data/00000-0-...-00001.parquet
default/ice_orders/metadata/snap-...-1-....avro
default/ice_orders/metadata/....-m0.avro
default/ice_orders/metadata/v2.metadata.json
default/ice_orders/metadata/version-hint.text

Read rows back

Query the table. CMF resolves and ships the same artifact, and the job reads the current snapshot:

confluent flink statement create ice-select \
  --url http://cmf:8080 \
  --environment test \
  --compute-pool iceberg-pool \
  --catalog _env_test \
  --database default \
  --sql "SELECT * FROM \`ice_orders\`;"

The query returns the three rows:

id  data
1   a
2   b
3   c

Clean up

Drop the table, then remove the pool and artifact:

confluent flink statement create ice-drop \
  --url http://cmf:8080 --environment test --compute-pool iceberg-pool \
  --catalog _env_test --database default \
  --sql "DROP TABLE IF EXISTS \`ice_orders\`;"

curl -X DELETE http://cmf:8080/cmf/api/v1/environments/test/compute-pools/iceberg-pool
curl -X DELETE http://cmf:8080/cmf/api/v1/environments/test/artifacts/iceberg-connector.jar

Dropping the table removes it from the environment catalog. Remove the Iceberg warehouse files from object storage to reclaim the space:

mc rm --recursive --force minio/flink/iceberg-warehouse/

Related content

• Custom Connectors and Formats in Confluent Manager for Apache Flink

• Manage Artifacts in Confluent Manager for Apache Flink

• Flink SQL Examples in Confluent Manager for Apache Flink