'Consume kafka topic JSON data to postgres using JDBC connector

We have a need to push the kafka topic JSON records to postGresSql database. the JSON are compliant to https://json-schema.org/draft-07/json-schema-release-notes.html

We tried to follow couple of examples online but could not help ourselves towards the solution.

connector properties

{
  "name": "person4",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "tasks.max": "1",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "topics": "personb",
    "connection.url": "jdbc:postgresql://localhost:5432/root?currentSchema=root",
    "connection.user": "root",
    "connection.password": "example",
    "insert.mode": "insert",
    "batch.size": "1",
    "delete.enabled": "false",
    "table.name.format": "person",
    "pk.mode": "none",
    "auto.create": "true",
    "auto.evolve": "true"
  }
}

sample data in topic

{ "name":"rob" }

schema for the data

{
  "$schema": "http://json-schema.org/draft-04/schema#",
  "properties": {
    "name": {
      "type": "string"
    }
  },
  "required": [
    "name"
  ],
  "type": "object"
}

The only expectation here is to get the JSON into one of the column in postgresql.

jsonpostgresqlapache-kafkaapache-kafka-connect


Solution 1:[1]

JDBC Sink requires a Schema.

To use JSONConverter for JSON data, it is not enough to have data adhere to any schema, rather the schema needs to be part of every record. The records should be of the form {"schema": ..., "payload":...}

https://www.confluent.io/blog/kafka-connect-deep-dive-converters-serialization-explained/

I'm not exactly sure on this, but I also think the record must have all the fields of the table. It's not possible to insert/update only one column, at least, without defining a primary key

Alternatively, you can setup the Confluent Schema Registry, and use their JSONSchemaConverter class, but this will require you to modify your producers to use the Schema Registry JSONSchema serializer classes, not send raw JSON events directly into the topic with no schema.

Sources

This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.

Source: Stack Overflow

Solution Source
Solution 1

Related Questions