Read CSV Files from Object Storage

This example reads a CSV file from S3-compatible object storage with the filesystem connector and the csv format. The cp-flink-sql image ships the csv format but not the filesystem connector, so you bundle both into one artifact and load it child-first.

For the artifact mechanism, see Custom Connectors and Formats in Confluent Manager for Apache Flink.

Important

This recipe reads files. Writing to object storage through the filesystem sink is not supported on the shipped image, because its S3 plugin has no recoverable writer.

Prerequisites

Complete the examples setup first. It provides the test environment, artifact management, and the MinIO object store with the mc alias this example uses. This example also needs:

• A CSV file in the object store at s3://flink/demo-data/products.csv. Create it and upload it with mc:

  cat > products.csv <<'EOF'
  p-100,Wireless Mouse,24.99
  p-101,Mechanical Keyboard,89.50
  p-102,USB-C Hub,39.00
  p-103,Laptop Stand,32.75
  EOF
  mc cp products.csv minio/flink/demo-data/products.csv
  

• The flink-connector-files and flink-csv JARs that match the Flink cluster version. The build below pulls flink-connector-files-1.19.1.jar and flink-csv-1.19.1.jar from Maven Central.

Bundle the connector and format into one JAR

The image ships flink-csv in its distribution but not flink-connector-files. If you upload the two JARs as separate artifacts, the job fails at runtime with NoClassDefFoundError: org/apache/flink/connector/file/table/factories/BulkReaderFormatFactory: the distribution’s flink-csv factory loads parent-first and cannot see the filesystem connector’s classes. Merge both JARs into one artifact so the CSV and file classes live together. Build the bundle with the Maven Shade plugin and its ServicesResourceTransformer, which merges the two factory service files instead of letting one overwrite the other. 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>filesystem-csv-bundle</artifactId>
  <version>1.0.0</version>
  <packaging>jar</packaging>

  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <flink.version>1.19.1</flink.version>
  </properties>

  <dependencies>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-connector-files</artifactId>
      <version>${flink.version}</version>
    </dependency>
    <dependency>
      <groupId>org.apache.flink</groupId>
      <artifactId>flink-csv</artifactId>
      <version>${flink.version}</version>
    </dependency>
  </dependencies>

  <build>
    <finalName>filesystem-csv-bundle</finalName>
    <plugins>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-shade-plugin</artifactId>
        <version>3.5.1</version>
        <executions>
          <execution>
            <phase>package</phase>
            <goals><goal>shade</goal></goals>
            <configuration>
              <!-- Bundle only these two JARs; their transitive Flink
                   dependencies are already on the pool image. -->
              <artifactSet>
                <includes>
                  <include>org.apache.flink:flink-connector-files</include>
                  <include>org.apache.flink:flink-csv</include>
                </includes>
              </artifactSet>
              <transformers>
                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
              </transformers>
            </configuration>
          </execution>
        </executions>
      </plugin>
    </plugins>
  </build>
</project>

Build it:

mvn package

The build writes target/filesystem-csv-bundle.jar with a single merged META-INF/services/org.apache.flink.table.factories.Factory listing the filesystem connector and both CSV factories.

Upload the bundle as a connector and as a format

CMF resolves the connector and the format independently, each in its own classloader, so upload the same bundle twice: once labeled connector with the filesystem identifier, and once labeled format with the csv identifier.

fs-connector.json:

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

csv-format.json:

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

Upload the bundle under each:

curl -X POST http://cmf:8080/cmf/api/v1/environments/test/artifacts \
  -F 'artifact=@fs-connector.json;type=application/json' \
  -F 'file=@filesystem-csv-bundle.jar'

curl -X POST http://cmf:8080/cmf/api/v1/environments/test/artifacts \
  -F 'artifact=@csv-format.json;type=application/json' \
  -F 'file=@filesystem-csv-bundle.jar'

Create a compute pool that loads the bundle child-first

By default Flink loads org.apache.flink. classes parent-first, so the distribution’s flink-csv factory shadows the bundle and dangles on the missing filesystem classes. Create a dedicated compute pool whose classloader.parent-first-patterns.default is the stock list with org.apache.flink. removed, so the bundle’s CSV and file classes resolve together, child-first. The pool also carries the S3 credentials the job uses to read the CSV. Save this as filesystem-pool.json:

{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "ComputePool",
  "metadata": { "name": "filesystem-pool" },
  "spec": {
    "type": "DEDICATED",
    "clusterSpec": {
      "flinkVersion": "v1_19",
      "image": "confluentinc/cp-flink-sql:1.19-cp11",
      "flinkConfiguration": {
        "taskmanager.numberOfTaskSlots": "4",
        "classloader.parent-first-patterns.default": "java.;scala.;com.esotericsoftware.kryo;org.apache.hadoop.;javax.annotation.;org.xml;javax.xml;org.apache.xerces;org.w3c;org.rocksdb.;org.slf4j;org.apache.log4j;org.apache.logging;org.apache.commons.logging;ch.qos.logback",
        "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 filesystem-pool.json \
  --environment test \
  --url http://cmf:8080

Removing org.apache.flink. from parent-first is a broad change, so keep it on this dedicated pool rather than the shared pool.

Create the table

Create a table on the CSV file, setting connector to filesystem and format to csv:

confluent flink statement create fs-create \
  --url http://cmf:8080 \
  --environment test \
  --compute-pool filesystem-pool \
  --catalog _env_test \
  --database default \
  --sql "CREATE TABLE \`products\` (\`id\` STRING, \`name\` STRING, \`price\` DOUBLE) WITH ('connector' = 'filesystem', 'path' = 's3://flink/demo-data/products.csv', 'format' = 'csv');"

Read the file

Run the query on the child-first pool:

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

The query returns the four rows from the CSV file:

id     name                 price
p-100  Wireless Mouse       24.99
p-101  Mechanical Keyboard  89.5
p-102  USB-C Hub            39.0
p-103  Laptop Stand         32.75

Clean up

Drop the table, then remove the pool and both artifacts:

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

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

Remove the CSV file from object storage:

mc rm minio/flink/demo-data/products.csv

Related content

• Custom Connectors and Formats in Confluent Manager for Apache Flink

• Round-Trip an Apache Iceberg Table

• Manage Artifacts in Confluent Manager for Apache Flink

• Flink SQL Examples in Confluent Manager for Apache Flink