CSV Format

Format: Serialization Schema Format: Deserialization Schema

The CSV format allows to read and write CSV data based on an CSV schema. Currently, the CSV schema is derived from table schema.

Dependencies

In order to use the CSV format the following dependencies are required for both projects using a build automation tool (such as Maven or SBT) and SQL Client with SQL JAR bundles.

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-csv</artifactId>
  <version>1.15.0</version>
</dependency>

How to create a table with CSV format

Here is an example to create a table using Kafka connector and CSV format.

CREATE TABLE user_behavior (
  user_id BIGINT,
  item_id BIGINT,
  category_id BIGINT,
  behavior STRING,
  ts TIMESTAMP(3)
) WITH (
 'connector' = 'kafka',
 'topic' = 'user_behavior',
 'properties.bootstrap.servers' = 'localhost:9092',
 'properties.group.id' = 'testGroup',
 'format' = 'csv',
 'csv.ignore-parse-errors' = 'true',
 'csv.allow-comments' = 'true'
)

Format Options

Data Type Mapping

Currently, the CSV schema is always derived from table schema. Explicitly defining an CSV schema is not supported yet.

Flink CSV format uses jackson databind API to parse and generate CSV string.

The following table lists the type mapping from Flink type to CSV type.