JSON format

The JSON format enables you to read and write JSON data. You must add the JSON dependency to your project and define the format type in CREATE table to JSON.

The expected JSON schema will be derived from the table schema by default. Specifying the JSON schema manually is not supported.

Maven dependency

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-json</artifactId>
  <version>1.19.1-csa1.13.1.0</version>
</dependency>

For more information about the JSON format, see the Apache Flink documentation.

The following example shows the Kafka connector with JSON data type:

CREATE TABLE source_table (
  c1	INT,
  c2	STRING,
  c3	DECIMAL(38,18),
  c4	TIMESTAMP(3),
) WITH (
  	'connector.type'    	 = 'kafka',
	'connector.version' 	 = 'universal',
	'connector.topic'   	 = 'input_topic',
	'connector.startup-mode' = 'latest-offset',
	'connector.properties.bootstrap.servers' = '<hostname>:<port>',
	'connector.properties.group.id' = 'test',
	'format.type' = 'json'
);