Kafka messages sink to Clickhouse

[omurov]

Hello all,

We are trying to get JSON messages from Kafka into Clickhouse, following the Source and Sink connectors we are getting some residual JSON fields from Kafka that we do not need in Clickhouse, we just need the JSON payload.
How can we extract this field only?
Example message:
{
"session_id": null,
"headers": {},
"message": {
"CLIENT_ID": "someid",
"SESSION_ID": "7017558492",
"SESSION_NUMBER": "1",
"LANGUAGE": "en",
"EVENT_NAME": "session",
"TIMESTAMP": "1628691008",
"JSON_VALUE": {
"application_action": "open",
"application_app_state": "starting",
"application_device_id": "somevalue"
}
},
"offset": 7,
"partition": 4,
"source_type": "kafka",
"timestamp": "2021-08-11T14:10:08.144Z",
"topic": "EVENTS"
}

We need only what is inside message:
{
"CLIENT_ID": "someid",
"SESSION_ID": "7017558492",
"SESSION_NUMBER": "1",
"LANGUAGE": "en",
"EVENT_NAME": "session",
"TIMESTAMP": "1628691008",
"JSON_VALUE": {
"application_action": "open",
"application_app_state": "starting",
"application_device_id": "somevalue"
}
We cannot find any kind of extract or transform to satisfy this.

Any help is appreciated,
Alex

[spencergilbert]

It looks like you could just do a remap transform with the following VRL source:

. = .message

Assuming .message is an object already, if it's just a string you'd need to parse it first.

[24igbories]

Thanks for the help, it actually extracts and returns the following :

{"message": {
"CLIENT_ID": "someid",
"SESSION_ID": "7017558492",
"SESSION_NUMBER": "1",
"LANGUAGE": "en",
"EVENT_NAME": "session",
"TIMESTAMP": "1628691008",
"JSON_VALUE": {
"application_action": "open",
"application_app_state": "starting",
"application_device_id": "somevalue"}
}

I would like to be able to extract just the content of message and removing message so that I just get the following :

{
"CLIENT_ID": "someid",
"SESSION_ID": "7017558492",
"SESSION_NUMBER": "1",
"LANGUAGE": "en",
"EVENT_NAME": "session",
"TIMESTAMP": "1628691008",
"JSON_VALUE": {
"application_action": "open",
"application_app_state": "starting",
"application_device_id": "somevalue"}
}

I have used the remap and tried to extract the payload of the message item, but I wasn't successful,
Do you have any ideas on how to proceed ?
Many thanks !

[spencergilbert]

💭 perhaps del(.message) is what you're looking for then?

. = del(.message)

. = .message works for me in the VRL REPL, could you share your complete config and we can see if there's something else happening in your configuration?

[24igbories]

I tried this " . = del(.message) " , it shows the same output in the vector console .
Please find attached my configuration :

config.txt

Here is the console output :
{"message":"{"CLIENT_ID":"POSTMAN-POC","GA_SESSION_ID":"ID","GA_SESSION_NUMBER":"1","LANGUAGE":"en","EVENT_NAME":"event","TIMESTAMP":"1628779656","JSON_VALUE":"{\"application_device_id\":\"POSTMAN-POC\",\"client_id\":\"POSTMAN-POC\",\"correlation_id\":\"correl_id\",\"event_batch_transport_time_epoch\":1628779657,\"event_trigger\":\"app_start\",\"source_type\":\"http\",\"timestamp\":1628779656,\"measurement_id\":\"XXXXXXXXXX\"}"}"}

I'm pulling the data from a kafka topic and sending it to clickhouse (and the console for debug purposes)

Thanks a lot for your help !

[jszwedko]

Hey @24igbories !

I think the issue you are seeing here is that the incoming message from the kafka source is serialized JSON. You'll need to parse it first, before remapping. You can do this like:

[transforms.kafka_transform]
  type = "remap"
  inputs = [ "kafka_logs" ]
  source = """
   . = parse_json!(.message)
   . = .message
  """
  drop_on_abort = true

[24igbories]

Thanks a lot for the suggestion,

Actually I tried it and it gave me a null message in the vector logs but I actually got the result I wanted from your code by just removing the the " . = .message" in source,
the pars_json on the message itself did the trick :

transforms.kafka_transform]
type = "remap"
inputs = [ "kafka_logs" ]
source = """
. = parse_json!(.message)
"""
drop_on_abort = true

I can see my messages flow into Clickhouse now,
Thanks a lot for the help !
I

[omurov]

For further development, is there a way for quick checks on VRL? Like running test in console? Would be great to have a test tool to speedup the development process. With kind regards, Alex

…

Sent from my iPhone

On 13 Aug 2021, at 09:21, 24igbories ***@***.***> wrote:  Thanks a lot for the suggestion, Actually I tried it and it gave me a null message in the vector logs but I actually got the result I wanted from your code by just removing the the " . = .message" in source, the pars_json on the message itself did the trick : transforms.kafka_transform] type = "remap" inputs = [ "kafka_logs" ] source = """ . = parse_json!(.message) """ drop_on_abort = true I can see my messages flow into Clickhouse now, Thanks a lot for the help ! I — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub, or unsubscribe. Triage notifications on the go with GitHub Mobile for iOS or Android.

[jszwedko]

Hey! There is. You can actually run vector vrl for a REPL. See vector vrl --help for documentation.