{"id":5857,"date":"2026-02-18T18:28:10","date_gmt":"2026-02-18T18:28:10","guid":{"rendered":"https:\/\/dalelane.co.uk\/blog\/?p=5857"},"modified":"2026-03-14T21:02:17","modified_gmt":"2026-03-14T21:02:17","slug":"processing-json-with-kafka-connect","status":"publish","type":"post","link":"https:\/\/dalelane.co.uk\/blog\/?p=5857","title":{"rendered":"Processing JSON with Kafka Connect"},"content":{"rendered":"

In this post, I’ll share examples of how to process JSON data in a Kafka Connect pipeline, and explain the schema format that Kafka uses to describe JSON events.<\/strong>\u00a0<\/p>\n

Using sink connectors<\/h3>\n

Kafka Connect sink connectors let you send the events on your Kafka topics\u00a0to external systems.\u00a0I’ve talked about this before<\/a>, but to recap the structure looks a bit like this:<\/p>\n

<\/p>\n

Imagine that you have this JSON event on a Kafka topic.\u00a0<\/p>\n

{\n    \"id\": 12345678,\n    \"message\": \"Hello World\",\n    \"isDemo\": true\n}<\/pre>\n
How should you configure Kafka Connect to send that somewhere?\u00a0<\/p>\n
It depends…<\/p>\n
<\/p>\n
Unstructured sink connectors<\/h3>\n
If you’re sending the event to a sink that doesn’t require structure, you could simply send use StringConverter<\/code>. The JSON event won’t be parsed or interpreted; it will just be sent as a string.\u00a0<\/p>\n
The simplest example of this is the demo file connector FileStreamSinkConnector<\/code>.<\/p>\n
<\/p>\n
With config like this, the Connector can receive each new event from the Kafka topic and\u00a0append it to a text file as a new (string) line.\u00a0<\/p>\n
value.converter=org.apache.kafka.connect.storage.StringConverter\nvalue.converter.schemas.enable=false\n\nconnector.class=FileStreamSink\nfile=mydemofile.jsonl<\/pre>\n
Structured sink connectors<\/h3>\n
What if you’re sending the event to a sink that does require structured data?<\/p>\n
An example of this would be a database connectors that inserts events into a PostgreSQL database.\u00a0<\/p>\n
<\/p>\n
If you’re sending the event to a sink that requires structure, you can use JsonConverter<\/code>.\u00a0<\/p>\n
There are other things that schemas can help with. For example, even if you’re using Kafka Connect with a sink that accepts unstructured data, if you want to use any transforms to process the events first, you’ll likely need to parse the raw JSON. Single Message Transforms are less useful when they only have raw string payloads to work with.<\/p>\n
<\/p>\n
Introducing schemas for JSON events<\/h4>\n
But<\/strong> to help with either of these, you need a way to tell JsonConverter about the structure of the data.\u00a0<\/p>\n
You might think that you could infer the structure from the event data, but that would just be guessing. If the structure is significant to the sink that receives the data, guesses are not enough – you won’t know about any missing optional values that have defaults, you won’t be able to distinguish between ambiguous numerical types, and so on. A schema is needed to fill in these missing gaps.\u00a0<\/p>\n
Kafka Connect uses its own JSON schema format, that is different<\/strong> from the IETF standard<\/a> we normally mean when we say “JSON schema”.\u00a0<\/p>\n
Demo<\/h3>\n
I’ll start with a simple example, and then step back and fill in some details.<\/p>\n
For this example, I’ll use a database sink connector<\/a> as an example of the sort of connector where data structure is essential.\u00a0<\/p>\n
<\/p>\n
Imagine that you have this JSON event on a Kafka topic that the connector should insert into a PostgreSQL database.\u00a0<\/p>\n
{\n    \"id\": \"00000000-0000-0000-0000-000000000000\",\n\n    \"example_integer_1\": 0,\n    \"example_integer_2\": 0,\n    \"example_integer_3\": 0,\n    \"example_integer_4\": 0,\n\n    \"example_decimal_1\": 0.0,\n    \"example_decimal_2\": 0.0,\n\n    \"example_boolflag\": false,\n\n    \"example_bytes\": \"\",\n\n    \"example_array_ints\": [ ],\n\n    \"example_temporal_1\": 0,\n    \"example_temporal_2\": 0,\n    \"example_temporal_3\": 0,\n\n    \"example_optional_1\": \"\",\n    \"example_optional_2\": \"\"\n}<\/pre>\n
A schema can describe each of the properties:<\/p>\n
{\n    \"type\": \"struct\",\n    \"fields\": [\n        {\n            \"field\": \"id\",\n            \"type\": \"string\",\n            \"optional\": false\n        },\n        {\n            \"field\": \"example_integer_1\",\n            \"type\": \"int8\",\n            \"optional\": false\n        },\n        {\n            \"field\": \"example_integer_2\",\n            \"type\": \"int16\",\n            \"optional\": false\n        },\n        {\n            \"field\": \"example_integer_3\",\n            \"type\": \"int32\",\n            \"optional\": false\n        },\n        {\n            \"field\": \"example_integer_4\",\n            \"type\": \"int64\",\n            \"optional\": false\n        },\n        {\n            \"field\": \"example_decimal_1\",\n            \"type\": \"float\",\n            \"optional\": false\n        },\n        {\n            \"field\": \"example_decimal_2\",\n            \"type\": \"double\",\n            \"optional\": false\n        },\n        {\n            \"field\": \"example_boolflag\",\n            \"type\": \"boolean\",\n            \"optional\": false\n        },\n        {\n            \"field\": \"example_bytes\",\n            \"type\": \"bytes\",\n            \"optional\": false\n        },\n        {\n            \"field\": \"example_array_ints\",\n            \"type\": \"array\",\n            \"items\": {\n                \"type\": \"int32\",\n                \"optional\": false\n            }\n        },\n        {\n            \"field\": \"example_temporal_1\",\n            \"type\": \"int32\",\n            \"name\": \"org.apache.kafka.connect.data.Date\",\n            \"optional\": false\n        },\n        {\n            \"field\": \"example_temporal_2\",\n            \"type\": \"int32\",\n            \"name\": \"org.apache.kafka.connect.data.Time\",\n            \"optional\": false\n        },\n        {\n            \"field\": \"example_temporal_3\",\n            \"type\": \"int64\",\n            \"name\": \"org.apache.kafka.connect.data.Timestamp\",\n            \"optional\": false\n        },\n        {\n            \"field\": \"example_optional_1\",\n            \"type\": \"string\",\n            \"optional\": true\n        },\n        {\n            \"field\": \"example_optional_2\",\n            \"type\": \"string\",\n            \"optional\": true,\n            \"default\": \"default_value_if_not_provided\"\n        }\n    ]\n}<\/pre>\n
You can set up your sink connector like this, using the JsonConverter<\/code> and the location of the schema.json file.<\/p>\n
value.converter=org.apache.kafka.connect.json.JsonConverter\nvalue.converter.schemas.enable=true\nvalue.converter.schema.content=${dirProvider:\/Users\/dalelane\/demos\/kafka-connect-json:schema.json}\n\nconnector.class=io.aiven.connect.jdbc.JdbcSinkConnector\n\n#\n# database connector config - not specific to the JSON converter\nconnection.url=jdbc:postgresql:\/\/localhost:5432\/dalelane\nconnection.user=dalelane\n# connection.password - my local dev database doesn't need a password, but your database probably does\ntable.name.format=public.demotable\ninsert.mode=upsert\nauto.create=true\npk.mode=record_value\npk.fields=id<\/pre>\n
Start up Connect, and you’ll see your PostgreSQL database has a new table called demotable.<\/p>\n
The JsonConverter has given the sink connector enough detail about the events on the topic for it to create a\u00a0table with the appropriate columns.<\/p>\n
Notice how the different column types are used, in a way that would be impossible to do from the event JSON alone.\u00a0<\/p>\n
dalelane=# \\d demotable\n                                         Table \"public.demotable\"\n       Column       |            Type             | Collation | Nullable |                Default\n--------------------+-----------------------------+-----------+----------+---------------------------------------\n id                 | text                        |           | not null |\n example_integer_1  | smallint                    |           | not null |\n example_integer_2  | smallint                    |           | not null |\n example_integer_3  | integer                     |           | not null |\n example_integer_4  | bigint                      |           | not null |\n example_decimal_1  | real                        |           | not null |\n example_decimal_2  | double precision            |           | not null |\n example_boolflag   | boolean                     |           | not null |\n example_bytes      | bytea                       |           | not null |\n example_array_ints | integer[]                   |           | not null |\n example_temporal_1 | date                        |           | not null |\n example_temporal_2 | time without time zone      |           | not null |\n example_temporal_3 | timestamp without time zone |           | not null |\n example_optional_1 | text                        |           |          |\n example_optional_2 | text                        |           |          | 'default_value_if_not_provided'::text\nIndexes:\n    \"demotable_pkey\" PRIMARY KEY, btree (id)\n<\/pre>\n
And a row representing the JSON event from the Kafka topic is there:<\/p>\n
dalelane=# select id, example_integer_1, example_integer_2, example_integer_3, example_integer_4 from demotable;\n                  id                  | example_integer_1 | example_integer_2 | example_integer_3 | example_integer_4\n--------------------------------------+-------------------+-------------------+-------------------+--------------------\n 00000000-0000-0000-0000-000000000000 |                 0 |                 0 |                 0 |                  0\n(1 row)\n\ndalelane=# select id, example_decimal_1, example_decimal_2 from demotable;\n                  id                  | example_decimal_1 | example_decimal_2\n--------------------------------------+-------------------+-------------------\n 00000000-0000-0000-0000-000000000000 |                 0 |                 0\n(1 row)\n\ndalelane=# select id, example_boolflag, example_bytes, example_array_ints from demotable;\n                  id                  | example_boolflag |    example_bytes     |  example_array_ints\n--------------------------------------+------------------+----------------------+-----------------------\n 00000000-0000-0000-0000-000000000000 | f                | \\x                   | {}\n(1 row)\n\ndalelane=# select id, example_temporal_1, example_temporal_2, example_temporal_3 from demotable;\n                  id                  | example_temporal_1 | example_temporal_2 | example_temporal_3\n--------------------------------------+--------------------+--------------------+---------------------\n 00000000-0000-0000-0000-000000000000 | 1970-01-01         | 00:00:00           | 1970-01-01 00:00:00\n(1 row)\n\ndalelane=# select id, example_optional_1, example_optional_2 from demotable;\n                  id                  | example_optional_1 |      example_optional_2\n--------------------------------------+--------------------+-------------------------------\n 00000000-0000-0000-0000-000000000000 |                    | default_value_if_not_provided\n(1 row)\n<\/pre>\n
To show another example, imagine you produce this second JSON event to the Kafka topic:<\/p>\n
{\n    \"example_decimal_1\": 3.1415927,\n    \"example_decimal_2\": 3.141592653589793,\n\n    \"example_boolflag\": true,\n\n    \"example_array_ints\": [ 2, 3, 5, 7, 11, 13, 17, 19 ],\n\n    \"example_integer_4\": 473892472947371246,\n    \"example_integer_3\": 1194727881,\n    \"example_integer_2\": 20472,\n    \"example_integer_1\": 12,\n\n    \"example_bytes\": \"ZGFsZS1sYW5l\",\n\n    \"example_temporal_1\": 3776,\n    \"example_temporal_3\": 1777885200000,\n    \"example_temporal_2\": 43200000,\n\n    \"id\": \"9D51DCBE-E2F1-47DA-B341-DAC8369C8451\"\n}<\/pre>\n
The connector will insert this new event data into the DB table:<\/p>\n
dalelane=# select id, example_integer_1, example_integer_2, example_integer_3, example_integer_4 from demotable;\n                  id                  | example_integer_1 | example_integer_2 | example_integer_3 | example_integer_4\n--------------------------------------+-------------------+-------------------+-------------------+--------------------\n 00000000-0000-0000-0000-000000000000 |                 0 |                 0 |                 0 |                  0\n 9D51DCBE-E2F1-47DA-B341-DAC8369C8451 |                12 |             20472 |        1194727881 | 473892472947371246\n(2 rows)\n\ndalelane=# select id, example_decimal_1, example_decimal_2 from demotable;\n                  id                  | example_decimal_1 | example_decimal_2\n--------------------------------------+-------------------+-------------------\n 00000000-0000-0000-0000-000000000000 |                 0 |                 0\n 9D51DCBE-E2F1-47DA-B341-DAC8369C8451 |         3.1415927 | 3.141592653589793\n(2 rows)\n\ndalelane=# select id, example_boolflag, example_bytes, example_array_ints from demotable;\n                  id                  | example_boolflag |    example_bytes     |  example_array_ints\n--------------------------------------+------------------+----------------------+-----------------------\n 00000000-0000-0000-0000-000000000000 | f                | \\x                   | {}\n 9D51DCBE-E2F1-47DA-B341-DAC8369C8451 | t                | \\x64616c652d6c616e65 | {2,3,5,7,11,13,17,19}\n(2 rows)\n\ndalelane=# select id, example_temporal_1, example_temporal_2, example_temporal_3 from demotable;\n                  id                  | example_temporal_1 | example_temporal_2 | example_temporal_3\n--------------------------------------+--------------------+--------------------+---------------------\n 00000000-0000-0000-0000-000000000000 | 1970-01-01         | 00:00:00           | 1970-01-01 00:00:00\n 9D51DCBE-E2F1-47DA-B341-DAC8369C8451 | 1980-05-04         | 12:00:00           | 2026-05-04 09:00:00\n(2 rows)\n\ndalelane=# select id, example_optional_1, example_optional_2 from demotable;\n                  id                  | example_optional_1 |      example_optional_2\n--------------------------------------+--------------------+-------------------------------\n 00000000-0000-0000-0000-000000000000 |                    | default_value_if_not_provided\n 9D51DCBE-E2F1-47DA-B341-DAC8369C8451 |                    | default_value_if_not_provided\n(2 rows)\n<\/pre>\n
Notice some of the benefits the schema brought, even beyond getting this to work at all.<\/p>\n
For example, the database engine added the correct default value for the example_optional_2 property that was missing from the Kafka event – as defined in the schema.\u00a0<\/p>\n
And notice that even though the ordering of the properties in the JSON payload was different in this second event, the correct values were inserted into the correct columns – because this is a structured event payload, not a single string.<\/p>\n
Writing a schema for JsonConverter<\/h3>\n
The example schema above is probably enough to give you an idea, as I intentionally included a lot of different types. I’ll summarise the basics below.<\/p>\n
Kafka Connect supports these primitive types:<\/p>\n