How to Create Stream from CSV Topic in ksqlDB

To create a stream from a topic containing records in CSV format, use the DELIMITED format. This is an example where the CSV records contain 3 different columns:

CREATE STREAM population_stream (
  id STRING,
  name STRING,
  age INT
) WITH (
  KAFKA_TOPIC = 'population',
  VALUE_FORMAT = 'DELIMITED',
  VALUE_DELIMITER = ','
);

The stream above is created using a topic named population, which looks like this:

ksql> print population;
Key format: ¯\_(Ä)_/¯ - no data processed
Value format: KAFKA_STRING
rowtime: 2023/01/26 11:30:10.071 Z, key: <null>, value: 1,Alice,35, partition: 0
rowtime: 2023/01/26 11:30:21.851 Z, key: <null>, value: 2,Bob,12, partition: 0
rowtime: 2023/01/26 11:30:42.483 Z, key: <null>, value: 3,Charlie,78, partition: 0

To parse values in the key, use KEY_FORMAT and KEY_DELIMITER instead.

If instead of commas, you use a different delimiter, set the delimiter to what you need. Remember it can only be one character.

To specify spaces or tabs as the delimiter, you need to pass SPACE or TAB as the delimiter value, instead of or \t.