# 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](../custom-connectors.md#cmf-sql-custom-connectors). #### 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](setup.md#cmf-sql-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`: ```bash 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`: ```xml 4.0.0 io.confluent.examples.flink filesystem-csv-bundle 1.0.0 jar UTF-8 1.19.1 org.apache.flink flink-connector-files ${flink.version} org.apache.flink flink-csv ${flink.version} filesystem-csv-bundle org.apache.maven.plugins maven-shade-plugin 3.5.1 package shade org.apache.flink:flink-connector-files org.apache.flink:flink-csv ``` Build it: ```bash 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 each connector and format identifier separately, to the artifact labeled with that type and identifier, and loads all the JARs a statement resolves together, not each in its own classloader. Upload the same bundle twice so that both identifiers resolve to it: once labeled `connector` with the `filesystem` identifier, and once labeled `format` with the `csv` identifier. `fs-connector.json`: ```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`: ```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: ```bash curl -X POST http://cmf:8080/cmf/api/v1/environments/test/artifacts \ -F 'artifact=@fs-connector.json;type=application/json' \ -F 'file=@target/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=@target/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`: ```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" } } } } } ``` ```bash 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`: ```bash 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: ```bash 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: ```text 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: ```bash 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: ```bash mc rm minio/flink/demo-data/products.csv ``` ## Related content - [Custom Connectors and Formats in Confluent Manager for Apache Flink](../custom-connectors.md#cmf-sql-custom-connectors) - [Round-Trip an Apache Iceberg Table](iceberg.md#cmf-sql-example-iceberg) - [Manage Artifacts in Confluent Manager for Apache Flink](../../../configure/artifacts.md#cmf-artifacts) - [Flink SQL Examples in Confluent Manager for Apache Flink](index.md#cmf-sql-examples)