How configure sink data from Oracle to Clickhouse, with debezium connector and clickhouse-kafka-connect

[Homicevich]

Hello!
I`m try synchronize a simple table from Oracle to Clickhouse.

Table structure in Oracle:

CREATE TABLE "TEST"."NEWTABLE"
( "COLUMN1" VARCHAR2(22)

)

Debezium is used to export data from Oracle.
Debezium configuration:

{
  "name": "orcl",
  "config": {
    "connector.class": "io.debezium.connector.oracle.OracleConnector",
    "tasks.max": "1",
    "database.hostname": "127.0.0.1",
    "database.port": "1521",
    "database.user": "dbuser",
    "database.password": "dbz",
    "database.dbname": "ORCL",
    "database.schema": "TEST",
    "database.server.name": "0d508efd4a5c",
    "table.include.list": "TEST.NEWTABLE",
    "topic.prefix": "dbz",
    "database.history.kafka.bootstrap.servers": "broker:9092",
    "database.history.kafka.topic": "schema-changes",
    "schema.history.internal.kafka.topic": "schema-changes.inventory",
    "schema.history.internal.kafka.bootstrap.servers": "broker:9092",
    "transforms": "flatten",
    "transforms.flatten.type": "org.apache.kafka.connect.transforms.Flatten$Value",
    "transforms.flatten.delimiter": "_"
  }
}

Table structure in Clickhouse:

CREATE TABLE click_import.NEWTABLE
(

 `after_COLUMN1` Nullable(String)
)
ENGINE = MergeTree
ORDER BY after_COLUMN1;

clickhouse - sink is used to import data into Clickhouse
Configuration:

{
 "name": "ch",
 "config": {
 "connector.class": "com.clickhouse.kafka.connect.ClickHouseSinkConnector",
 "tasks.max": "1",
 "consumer.override.max.poll.records": "5000",
 "consumer.override.max.partition.fetch.bytes": "5242880",
 "database": "click_import",
 "errors.retry.timeout": "60",
 "exactlyOnce": "false",
 "hostname": "127.0.0.1",
 "port": "8123",
 "ssl": "false",
 "jdbcConnectionProperties": "?ssl=false",
 "username": "click_import",
 "password": "111",
"topics": "dbz.TEST.NEWTABLE",
"topic2TableMap": "dbz.TEST.NEWTABLE=NEWTABLE",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "true",
"clickhouseSettings": "",
"suppressTableExistenceException": "true"
}
}

When adding data to a table, Oracle Debezium generates the following record in Kafka:

{
	"schema": {
		"type": "struct",
		"fields": [
			{
				"type": "string",
				"optional": true,
				"field": "before_COLUMN1"
			},
			{
				"type": "string",
				"optional": true,
				"field": "after_COLUMN1"
			},
			{
				"type": "string",
				"optional": false,
				"field": "source_version"
			},
			{
				"type": "string",
				"optional": false,
				"field": "source_connector"
			},
			{
				"type": "string",
				"optional": false,
				"field": "source_name"
			},
			{
				"type": "int64",
				"optional": false,
				"field": "source_ts_ms"
			},
			{
				"type": "string",
				"optional": true,
				"name": "io.debezium.data.Enum",
				"version": 1,
				"parameters": {
					"allowed": "true,last,false,incremental"
				},
				"default": "false",
				"field": "source_snapshot"
			},
			{
				"type": "string",
				"optional": false,
				"field": "source_db"
			},
			{
				"type": "string",
				"optional": true,
				"field": "source_sequence"
			},
			{
				"type": "int64",
				"optional": true,
				"field": "source_ts_us"
			},
			{
				"type": "int64",
				"optional": true,
				"field": "source_ts_ns"
			},
			{
				"type": "string",
				"optional": false,
				"field": "source_schema"
			},
			{
				"type": "string",
				"optional": false,
				"field": "source_table"
			},
			{
				"type": "string",
				"optional": true,
				"field": "source_txId"
			},
			{
				"type": "string",
				"optional": true,
				"field": "source_scn"
			},
			{
				"type": "string",
				"optional": true,
				"field": "source_commit_scn"
			},
			{
				"type": "string",
				"optional": true,
				"field": "source_lcr_position"
			},
			{
				"type": "string",
				"optional": true,
				"field": "source_rs_id"
			},
			{
				"type": "int64",
				"optional": true,
				"field": "source_ssn"
			},
			{
				"type": "int32",
				"optional": true,
				"field": "source_redo_thread"
			},
			{
				"type": "string",
				"optional": true,
				"field": "source_user_name"
			},
			{
				"type": "string",
				"optional": true,
				"field": "source_redo_sql"
			},
			{
				"type": "string",
				"optional": true,
				"field": "source_row_id"
			},
			{
				"type": "string",
				"optional": true,
				"field": "transaction_id"
			},
			{
				"type": "int64",
				"optional": true,
				"field": "transaction_total_order"
			},
			{
				"type": "int64",
				"optional": true,
				"field": "transaction_data_collection_order"
			},
			{
				"type": "string",
				"optional": false,
				"field": "op"
			},
			{
				"type": "int64",
				"optional": true,
				"field": "ts_ms"
			},
			{
				"type": "int64",
				"optional": true,
				"field": "ts_us"
			},
			{
				"type": "int64",
				"optional": true,
				"field": "ts_ns"
			}
		],
		"optional": false,
		"name": "dbz.TEST.NEWTABLE.Envelope",
		"version": 2
	},
	"payload": {
		"before_COLUMN1": null,
		"after_COLUMN1": "4445454545",
		"source_version": "3.0.4.Final",
		"source_connector": "oracle",
		"source_name": "dbz",
		"source_ts_ms": 1734949222000,
		"source_snapshot": "false",
		"source_db": "FTS",
		"source_sequence": null,
		"source_ts_us": 1734949222000000,
		"source_ts_ns": 1734949222000000000,
		"source_schema": "TEST",
		"source_table": "NEWTABLE",
		"source_txId": "0a000400092c0200",
		"source_scn": "7555550017308",
		"source_commit_scn": "7555550017309",
		"source_lcr_position": null,
		"source_rs_id": "0x001154.00025b0d.0010",
		"source_ssn": 0,
		"source_redo_thread": 1,
		"source_user_name": "TEST",
		"source_redo_sql": null,
		"source_row_id": "AABXY/AAHAAACwtAAB",
		"transaction_id": null,
		"transaction_total_order": null,
		"transaction_data_collection_order": null,
		"op": "c",
		"ts_ms": 1734949226085,
		"ts_us": 1734949226085622,
		"ts_ns": 1734949226085622067
	}
}

At the moment, 2 problems have appeared:

1. For normal operation, it was necessary to create a column in Clickhouse with the name after_COLUMN1, otherwise the data was not added to the table. Previously, I tried to use a column with the name COLUMN1 in this case NULL was added to the table.
2. When changing data in a table in Oracle, they do not change in Clickhouse. The data is simply added as new.

Pls Tell me which config to use to solve these problems? especially the second one

I`m use:
Clickhouse: 24.11
Clickhouse connect sink : 1.26

[Paultagoras]

Hi and apologies for the delay - was away for the holidays 😅

• Yes the tables are flattened as before_ and after_ - that's the expected behavior. You could experiment with trying without flatten (or using the String converter rather than JSON) but I don't know what the results would be off the top.
• When folks want data that gets replaced (rather than added as new) we generally see https://clickhouse.com/docs/en/guides/replacing-merge-tree as the better table engine option.