I'm using debezium with a simple flat table structure, what is the easiest way to persist this data in clickhouse?

[alberttwong]

I'm using debezium with a simple flat table structure (4 fields), what is the easiest way to persist this data in clickhouse?

Since the kafka connector doesn't create a table. Here's the table create command.

CREATE TABLE default.customer
(
	`id` UInt64,
	`first_name` String,
	`last_name` String,
        `email` String

)
PRIMARY KEY (id)

connect config kafka_connect_clickhouse.json

{
  "name": "clickhouse-connect",
  "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": "default",
    "errors.retry.timeout": "60",
    "exactlyOnce": "false",
    "hostname": "sfmtp3rq7x.us-west-2.aws.clickhouse.cloud",
    "port": "8443",
    "ssl": "true",
    "jdbcConnectionProperties": "?ssl=true&sslmode=strict",
    "username": "default",
    "password": "XXXXXXX",
    "topics": "dbserver1.inventory.customers",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "topic2TableMap": "dbserver1.inventory.customers=customer",
    "clickhouseSettings": ""
  }
}

data

{
	"schema": {
		"type": "struct",
		"fields": [
			{
				"type": "struct",
				"fields": [
					{
						"type": "int32",
						"optional": false,
						"default": 0,
						"field": "id"
					},
					{
						"type": "string",
						"optional": false,
						"field": "first_name"
					},
					{
						"type": "string",
						"optional": false,
						"field": "last_name"
					},
					{
						"type": "string",
						"optional": false,
						"field": "email"
					}
				],
				"optional": true,
				"name": "dbserver1.inventory.customers.Value",
				"field": "before"
			},
			{
				"type": "struct",
				"fields": [
					{
						"type": "int32",
						"optional": false,
						"default": 0,
						"field": "id"
					},
					{
						"type": "string",
						"optional": false,
						"field": "first_name"
					},
					{
						"type": "string",
						"optional": false,
						"field": "last_name"
					},
					{
						"type": "string",
						"optional": false,
						"field": "email"
					}
				],
				"optional": true,
				"name": "dbserver1.inventory.customers.Value",
				"field": "after"
			},
			{
				"type": "struct",
				"fields": [
					{
						"type": "string",
						"optional": false,
						"field": "version"
					},
					{
						"type": "string",
						"optional": false,
						"field": "connector"
					},
					{
						"type": "string",
						"optional": false,
						"field": "name"
					},
					{
						"type": "int64",
						"optional": false,
						"field": "ts_ms"
					},
					{
						"type": "string",
						"optional": true,
						"name": "io.debezium.data.Enum",
						"version": 1,
						"parameters": {
							"allowed": "true,last,false,incremental"
						},
						"default": "false",
						"field": "snapshot"
					},
					{
						"type": "string",
						"optional": false,
						"field": "db"
					},
					{
						"type": "string",
						"optional": true,
						"field": "sequence"
					},
					{
						"type": "int64",
						"optional": true,
						"field": "ts_us"
					},
					{
						"type": "int64",
						"optional": true,
						"field": "ts_ns"
					},
					{
						"type": "string",
						"optional": false,
						"field": "schema"
					},
					{
						"type": "string",
						"optional": false,
						"field": "table"
					},
					{
						"type": "int64",
						"optional": true,
						"field": "txId"
					},
					{
						"type": "int64",
						"optional": true,
						"field": "lsn"
					},
					{
						"type": "int64",
						"optional": true,
						"field": "xmin"
					}
				],
				"optional": false,
				"name": "io.debezium.connector.postgresql.Source",
				"field": "source"
			},
			{
				"type": "struct",
				"fields": [
					{
						"type": "string",
						"optional": false,
						"field": "id"
					},
					{
						"type": "int64",
						"optional": false,
						"field": "total_order"
					},
					{
						"type": "int64",
						"optional": false,
						"field": "data_collection_order"
					}
				],
				"optional": true,
				"name": "event.block",
				"version": 1,
				"field": "transaction"
			},
			{
				"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": "dbserver1.inventory.customers.Envelope",
		"version": 2
	},
	"payload": {
		"before": null,
		"after": {
			"id": 1004,
			"first_name": "Anne",
			"last_name": "Kretchmar",
			"email": "annek@noanswer.org"
		},
		"source": {
			"version": "2.7.0.Final",
			"connector": "postgresql",
			"name": "dbserver1",
			"ts_ms": 1721680031481,
			"snapshot": "last_in_data_collection",
			"db": "postgres",
			"sequence": "[null,\"34527904\"]",
			"ts_us": 1721680031481890,
			"ts_ns": 1721680031481890000,
			"schema": "inventory",
			"table": "customers",
			"txId": 772,
			"lsn": 34527904,
			"xmin": null
		},
		"transaction": null,
		"op": "r",
		"ts_ms": 1721680031535,
		"ts_us": 1721680031535633,
		"ts_ns": 1721680031535633700
	}
}

[Paultagoras]

If you want everything to be in a singular table (which if they're all in the same topic, makes sense) I would just create a table with all of those fields present:

CREATE TABLE default.customer
(
after.id UInt64,
after.first_name String,
after.last_name String,
after.email String,
...
)
PRIMARY KEY (id)

[Paultagoras]

I would still recommend flattening the data though using the SMT

[Paultagoras]

One other option that might help - we added a flag to bypass some of the internal logic, if you run into data validation issues. It's bypassRowBinary as described in https://clickhouse.com/docs/en/integrations/kafka/clickhouse-kafka-connect-sink#configuration-options

[alberttwong]

Sure.. I can fatten the data through SMT.... then the next problem is table creation on the CH side when new kafka message appear.

[Paultagoras]

I'm confused - you only need to create the table initially, not for each message. Are the messages wildly different from each other?

[alberttwong]

I'm confused - you only need to create the table initially, not for each message. Are the messages wildly different from each other?

Sorry. I have a bunch of tables so it's helpful if a new topic just appears and the table in ch is created automatically.