How to push unstructured JSON data with Keys into a clickhouse table using this connector?

[sidharthramesh]

I have a Kafka topic with the following properties:

• Keys are always strings and represent the primary key
• Values are unstructured JSON where the keys and values of the JSON cannot be known in advance.

I want to push this data into Clickhouse for further analysis using the Clickhouse Kafka sink connector. However, I can't find the format for the target table (named the same as the topic). I'd ideally like a ReplacingMergeTree with the keys in the Kafka topic acting as the primary key, and each new message in the topic acting as a different version. And the value of the Kafka message should be inserted as a JSON / string so that it can be queried.

Can anyone help me with this scenario?

[Paultagoras]

I'd ideally like a ReplacingMergeTree with the keys in the Kafka topic acting as the primary key, and each new message in the topic acting as a different version. And the value of the Kafka message should be inserted as a JSON / string so that it can be queried.

What you're probably looking for is a message transform!

• Hoist Field, which should wrap the original message in a new field that you can set
• Insert Field, which then lets you put the key into the message.

I'm unable to test it this moment, but that SHOULD solve your issue. Hope it helps!

[Paultagoras]

If it's not a perfect fix, it may still get you most of the way there (and then require a bit of iteration)

[sidharthramesh]

Hey @Paultagoras, thanks a lot for your answer!

I'm very close to this setup working. I've gotten the values into Clickhouse, but no luck with being able to get the Kafka message's Key into the Value. InsertField SMT does not seem to support inserting the key into the message. I tried this configuration (as suggested by ChatGPT):

      transforms.InsertField.type: "org.apache.kafka.connect.transforms.InsertField$Value"
      transforms.InsertField.static.field: "key"
      transforms.InsertField.static.value: "$$Key"

However, this only inserts the static $$Key into the table:

This is how I created the table in Clickhouse:

CREATE TABLE "topic-name" (`key` String, `value` String, `insertTime` DateTime64(3) 
DEFAULT now()) Engine = MergeTree ORDER BY key;

I've also tried hoisting the key field, but it hoists it into the key itself - not very useful since the Clickhouse connector seems to only be looking at the value of a message.

Is there any other way to insert the key of a message into the value, so that Clickhouse can ingest the same?

[Paultagoras]

So I've been looking into the transformers available again to see if there's a better option - while apparently custom transformers are a thing, the out-of-the-box options don't seem to support it.

@mzitnik Do you know of a way to fold in the key into the message value?

[Paultagoras]

So after discussing with @mzitnik it seems that Custom Transforms are your best option to implement this - it supports what you're trying to achieve and it also allows you to reuse it multiple places.

Hope that helps!