Splitting json message based on the matching field name

[sesa657547]

Hi,

I am new to benthos and on the process of learning it. I have got a situation where I need to split the single payload into multiple messages based on the field names and send these messaged to different kafka topics. My payload looks like this:

[
{
"dId": "kjglsjglfkgfdkgfhjfhdsj",
"d1": "fhkfjsaf",
"d2": {
"dc1": 8,
"dc2": 9
},
"timestamp": "2020-09-03",
"A": [
{
"aId": "jflksflskgskg",
"af1": "kkfls",
"af2": {
"ac1": 6,
"ac2": 2
},
"af3": 1,

      "af4": "00",
      "af5": 4
    }
  ],
  "B": [
    {
      "bId": "jgljgldjgs",
      "bf1": 6,
      "bf2": {
        "bc1": 4,
        "bc2": 2
      }
    }
  ],
  "C": [
    {
      "cId": "njfglsjg",
      "cf1": 3,
      "cf2": "jfgg"
    }
  ]
}

]

I want to send A array data to one kafka topics and similarly for B and C.
Thanks
M

[mihaitodor]

Hey @sesa657547 👋 Here's how I'd do it:

input:
  generate:
    count: 1
    mapping: |
      root = [
        {
          "dId": "kjglsjglfkgfdkgfhjfhdsj",
          "d1": "fhkfjsaf",
          "d2": {
            "dc1": 8,
            "dc2": 9
          },
          "timestamp": "2020-09-03",
          "A": [
            {
              "aId": "jflksflskgskg",
              "af1": "kkfls",
              "af2": {
                "ac1": 6,
                "ac2": 2
              },
              "af3": 1,
              "af4": "00",
              "af5": 4
            }
          ],
          "B": [
            {
              "bId": "jgljgldjgs",
              "bf1": 6,
              "bf2": {
                "bc1": 4,
                "bc2": 2
              }
            }
          ],
          "C": [
            {
              "cId": "njfglsjg",
              "cf1": 3,
              "cf2": "jfgg"
            }
          ]
        }
      ]
  processors:
    - mapping: |
        root = this.0.with("A", "B", "C").key_values()
    - unarchive:
        format: json_array
    - mapping: |
        meta key = this.key

        root = this.value

output:
  kafka:
    # ...
    topic: ${! @key }
    # ...

I used the generate input to show you how you can test this payload as a standalone example.

If A, B and C aren't the actual topic names, you can massage them further using bloblang inside ${! ... }, or if you just need to prepend / append some string to them, you can also do stuff like foo-${! @key }-bar.

[sesa657547]

Thanks @mihaitodor , its giving me error like this: cannot unmarshal !!seq into pipeline.Config

[mihaitodor]

You'll have to show the full config you're running. That happens when you have an indentation issue or when you're not placing yaml nodes in the right place.

[sesa657547]

Hey, thank you. i re-checked my config and it actually worked. But I have one more question. If I need to add the below data: {
"dId": "kjglsjglfkgfdkgfhjfhdsj",
"d1": "fhkfjsaf",
"d2": {
"dc1": 8,
"dc2": 9
},
"timestamp": "2020-09-03"
to each newly created message payload , how can do it? I want this data to be added to every message sent to the kafka topics.

Really appreciate your help.
Thanks

[mihaitodor]

You're welcome! See this updated config:

input:
  generate:
    count: 1
    mapping: |
      root = [
        {
          "dId": "kjglsjglfkgfdkgfhjfhdsj",
          "d1": "fhkfjsaf",
          "d2": {
            "dc1": 8,
            "dc2": 9
          },
          "timestamp": "2020-09-03",
          "A": [
            {
              "aId": "jflksflskgskg",
              "af1": "kkfls",
              "af2": {
                "ac1": 6,
                "ac2": 2
              },
              "af3": 1,
              "af4": "00",
              "af5": 4
            }
          ],
          "B": [
            {
              "bId": "jgljgldjgs",
              "bf1": 6,
              "bf2": {
                "bc1": 4,
                "bc2": 2
              }
            }
          ],
          "C": [
            {
              "cId": "njfglsjg",
              "cf1": 3,
              "cf2": "jfgg"
            }
          ]
        }
      ]
  processors:
    - mapping: |
        meta extraFields = this.0.without("A", "B", "C")
        root = this.0.with("A", "B", "C").key_values()
    - unarchive:
        format: json_array
    - mapping: |
        meta key = this.key

        root = this.value.0.merge(@extraFields)

[Jeffail]

Moving to discussion (#2026)

[sesa657547]

Sorry about that. It would be great if you can help with any idea to resolve the problem. Thanks