Parse JSON Data in Flink Kafka, OSS & Elasticsearch Connectors - Realtime Compute for Apache Flink

The JSON format reads and writes JSON data. The JSON structure is automatically inferred from the table schema. Realtime Compute for Apache Flink uses the jackson databind API to parse and generate JSON data.

The following connectors support the JSON format:

Example

The following example creates a Kafka table that reads and writes data in JSON format.

CREATE TABLE Orders (
  orderId    INT,
  product    STRING,
  orderInfo  MAP<STRING, STRING>,
  orderTime  TIMESTAMP(3),
  WATERMARK FOR orderTime AS orderTime - INTERVAL '5' SECOND
) WITH (
  'connector'                  = 'kafka',
  'topic'                      = 'test-topic',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format'                     = 'json',
  'json.fail-on-missing-field' = 'false',
  'json.ignore-parse-errors'   = 'true'
);

Parameters

Parameter	Required	Default	Type	Description
`format`	Yes	(none)	STRING	The format to use. Set to `json`.
`json.fail-on-missing-field`	No	false	BOOLEAN	Specifies whether to skip a field or row when a field to be parsed is missing. `true`: skip the field or row. `false`: return an error and fail the deployment.
`json.ignore-parse-errors`	No	false	BOOLEAN	Specifies whether to skip a field or row when parsing fails. `true`: skip the field or row. `false`: return an error and fail the deployment.
`json.timestamp-format.standard`	No	SQL	STRING	The timestamp format for input and output. `SQL`: `yyyy-MM-dd HH:mm:ss.s{precision}` (for example, `2020-12-30 12:13:14.123`). `ISO-8601`: `yyyy-MM-ddTHH:mm:ss.s{precision}` (for example, `2020-12-30T12:13:14.123`).
`json.map-null-key.mode`	No	FAIL	STRING	How to handle null keys in a map. `FAIL`: return an error. `DROP`: discard the entry. `LITERAL`: replace the null key with the string specified by `json.map-null-key.literal`.
`json.map-null-key.literal`	No	null	STRING	The string constant used to replace null map keys when `json.map-null-key.mode` is set to `LITERAL`.
`json.encode.decimal-as-plain-number`	No	false	BOOLEAN	Specifies whether to encode DECIMAL values in plain notation. `true`: use plain notation (for example, `0.000000027`). `false`: use scientific notation (for example, `2.7E-8`).
`json.write-null-properties`	No	true	BOOLEAN	Specifies whether to write null column values to the JSON output. `true`: write null values. `false`: omit null fields. Requires Ververica Runtime (VVR) 8.0.6 or later.

Data type mappings

Flink SQL type	JSON type
CHAR / VARCHAR / STRING	STRING
BOOLEAN	BOOLEAN
BINARY / VARBINARY	STRING (base64-encoded)
DECIMAL	NUMBER
TINYINT	NUMBER
SMALLINT	NUMBER
INT	NUMBER
BIGINT	NUMBER
FLOAT	NUMBER
DOUBLE	NUMBER
DATE	STRING (`date` format)
TIME	STRING (`time` format)
TIMESTAMP	STRING (`date-time` format)
TIMESTAMP_WITH_LOCAL_TIME_ZONE	STRING (`date-time` format, UTC)
INTERVAL	NUMBER
ARRAY	ARRAY
MAP / MULTISET	OBJECT
ROW	OBJECT

Limitations

Data written to OSS through the JSON format cannot be saved as JSON files. For details, see FLINK-30635.