# Deserialization of results in the Output section

**URL:** <https://discuss.elastic.co/t/deserialization-of-results-in-the-output-section/298634>\
**Category:** Logstash\
**Created:** [March 2, 2022, 3:06pm UTC](https://discuss.elastic.co/t/deserialization-of-results-in-the-output-section/298634 "2022-03-02T15:06:34Z")\
**Posts on this page:** 8\
**Page:** 1

<div class="post-metadata">

**Author:** ![puneet\_makhija](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/puneet_makhija/32/100881_2.png) [@puneet\_makhija](https://discuss.elastic.co/u/puneet_makhija)\
**Post date:** [March 2, 2022, 3:06pm UTC](https://discuss.elastic.co/t/deserialization-of-results-in-the-output-section/298634/1 "2022-03-02T15:06:34Z")

</div>

@DavidTurner hope you are doing well can you please assist me into this one , I am stuck in this one  
I successfully implemented LogStash, here below is my logstash.conf code which is working perfectly

```auto
input {
  kafka {
        bootstrap_servers => "http://localhost:9092"
        topics => ["tracking"]
        codec => avro {
            schema_uri => "/home/abc/python/working/divolte-collector-with-apache-kafka/divolte-collector-0.9.0/conf/MyEventRecord.avsc"
        }
        value_deserializer_class => "org.apache.kafka.common.serialization.ByteArrayDeserializer"
    }
}

filter {
    if [eventType] == 'view' {
        prune {
            blacklist_names => ["^purchase___.*"]
        }
        ruby {
            code => "prefix = 'view___'
            event.to_hash.each{|k,v|
            if (!k.start_with?('@')) && (k.include?(prefix))
                event.set(k.split(prefix).last, v)
                event.remove(k)
            end
            }"
        }

    }

    if [eventType] == 'purchase' {
        prune {
            blacklist_names => ["^view___.*"]
        }
        ruby {
            code => "prefix = 'purchase___'
            event.to_hash.each{|k,v|
            if (!k.start_with?('@')) && (k.include?(prefix))
                event.set(k.split(prefix).last, v)
                event.remove(k)
            end
            }"
        }

    }
}

output {
    stdout {
        codec => rubydebug
    }

    if [eventType] == 'view' {
        kafka {
            bootstrap_servers => "http://localhost:9092"
            topic_id => 'view'
        }
    }
    else if [eventType] == 'purchase' {
        kafka {
            bootstrap_servers => "http://localhost:9092"
            topic_id => 'purchase'
        }
    }
    else {
        kafka {
            bootstrap_servers => "http://localhost:9092"
            topic_id => 'unrecognized'
        }

    }
}

```

here in the input section if you see we applied value\_deserializer\_class just to deserialize the result, and when in the consumer side in my Python code I am unable to deserialize this one , I tried various strategies but unable to deserialize this one.

```auto
consumer = KafkaConsumer(bootstrap_servers="localhost:9092", auto_offset_reset="earliest", value_deserializer=lambda m: json.loads(m).decode('utf-8'))

consumer.subscribe(['purchase'])
for message in consumer:
      print(message)

here I am getting JSONDecodeError Extra Data line 1 column 5 (char 4)

```

I removed decode() as well but this one also giving the same error

`consumer = KafkaConsumer(bootstrap_servers="localhost:9092", auto_offset_reset="earliest", value_deserializer=lambda m: json.loads(m))`

But what I see in the Logstash automatic shell this will giving the exact json result which I want , in the Consumer Python application  
{  
"order\_id"=\> "1",  
"session\_id"=\> "q:2332321sqknds",  
"unit\_price": "100",  
"tax": 10",  
"additional\_cost": ""

.....

}

so through this I tried one more thing I applied , in the output section

if [eventType] == 'purchase' {  
kafka {  
bootstrap\_servers =\> "[http://localhost:9092](http://localhost:9092)"  
topic\_id =\> 'purchase'  
value\_deserializer\_class =\> "org.apache.kafka.common.serialization.ByteArrayDeserializer"  
}  
}

But here this one is giving "value\_deserializer\_class" is not allowed in Kafka , even though I am using in **input section** value\_deserializer\_class

---

<div class="post-metadata">

**Author:** ![DavidTurner](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/davidturner/32/22453_2.png) [@DavidTurner](https://discuss.elastic.co/u/DavidTurner)\
**Post date:** [March 2, 2022, 3:18pm UTC](https://discuss.elastic.co/t/deserialization-of-results-in-the-output-section/298634/2 "2022-03-02T15:18:51Z")

</div>

It's quite impolite to ping random people who aren't already involved in your conversation. Please don't do this again. We're all volunteers here, we cannot address every question.

---

<div class="post-metadata">

**Author:** ![puneet\_makhija](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/puneet_makhija/32/100881_2.png) [@puneet\_makhija](https://discuss.elastic.co/u/puneet_makhija)\
**Post date:** [March 2, 2022, 3:19pm UTC](https://discuss.elastic.co/t/deserialization-of-results-in-the-output-section/298634/3 "2022-03-02T15:19:51Z")

</div>

> [@DavidTurner](#):
>
> It's quite impolite to ping random people who aren't already involved in your conversation. Please don't do this again. We're all volunteers here, we cannot address every question.

ok sorry

---

<div class="post-metadata">

**Author:** ![leandrojmp](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/leandrojmp/32/107231_2.png) [@leandrojmp](https://discuss.elastic.co/u/leandrojmp)\
**Post date:** [March 2, 2022, 4:18pm UTC](https://discuss.elastic.co/t/deserialization-of-results-in-the-output-section/298634/4 "2022-03-02T16:18:55Z")

</div>

You do not have a `value_deserializer_class` setting in the kafka output, only in the input.

In the output you have a [`value_serializer_class`](https://www.elastic.co/guide/en/logstash/current/plugins-outputs-kafka.html#plugins-outputs-kafka-value_serializer).

Can you give more context about what you expect to be in the output? What it should look like?

---

<div class="post-metadata">

**Author:** ![puneet\_makhija](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/puneet_makhija/32/100881_2.png) [@puneet\_makhija](https://discuss.elastic.co/u/puneet_makhija)\
**Post date:** [March 2, 2022, 4:58pm UTC](https://discuss.elastic.co/t/deserialization-of-results-in-the-output-section/298634/5 "2022-03-02T16:58:58Z")

</div>

@leandrojmp thanks for your response  
basically I want the object {"eventType": "purchase", "purchase\_\_\_order\_id": "123XC", purchase\_\_\_product\_price: "99", "view\_\_product\_id": ""}

which this logstash get into the input section and after I applied Ruby filter to remove some of the fields (event type is purchase =\> then remove view fields)  
(event type is view =\> then remove purchase fields)

After Applying filter  
{"eventType": "purchase", "purchase\_\_\_order\_id": "123XC", purchase\_\_\_product\_price: "99"}  
The object is looking something like this and this object I want to be easily accessible by my consumer application  
but my consumer application getting something like this  
**2022-03-22%{host}%{message}**

---

<div class="post-metadata">

**Author:** ![puneet\_makhija](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/puneet_makhija/32/100881_2.png) [@puneet\_makhija](https://discuss.elastic.co/u/puneet_makhija)\
**Post date:** [March 2, 2022, 5:49pm UTC](https://discuss.elastic.co/t/deserialization-of-results-in-the-output-section/298634/6 "2022-03-02T17:49:46Z")

</div>

In logstash running automatic configuration I am getting in terminal are the perfect results

**On view:**

{  
"created\_at" =\> 1646241664540,  
"session\_id" =\> "0:l09gy8t0:deZazkdX\_mux~VSA06rEr812XXnzlpBV",  
"customer\_id" =\> "22",  
"@version" =\> "1",  
"product\_id" =\> "34",  
"@timestamp" =\> 2022-03-02T17:21:04.554Z,  
"eventType" =\> "view"  
}

**on Purchase**

{  
"created\_at" =\> 1646237126471,  
"tax" =\> "5",  
"unit\_total\_cost" =\> "",  
"quantity" =\> "",  
"unit\_price" =\> "20",  
"session\_id" =\> "0:l09gy8t0:deZazkdX\_mux~VSA06rEr812XXnzlpBV",  
"customer\_id" =\> "282",  
"promo\_id" =\> "",  
"@version" =\> "1",  
"gross\_spend\_amount" =\> "",  
"product\_id" =\> "181",  
"@timestamp" =\> 2022-03-02T16:05:26.489Z,  
"order\_id" =\> "CC2",  
"eventType" =\> "purchase",  
"other\_cost" =\> "10",  
"discount" =\> "",  
"net\_spend\_amount" =\> "35"  
}

but in Kafka consumer

for message in consumer:  
print(message.value)  
#b'2022-03-02T17:36:24.056Z %{host} %{message}

**Behind the scene message contains:**

ConsumerRecord(topic='purchase', partition=0, offset=31, timestamp=1646242584162, timestamp\_type=0, key=None, value=b'2022-03-02T17:36:24.056Z %{host} %{message}', headers=, checksum=3789492333, serialized\_key\_size=-1, serialized\_value\_size=43, serialized\_header\_size=-1)

---

<div class="post-metadata">

**Author:** ![puneet\_makhija](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/puneet_makhija/32/100881_2.png) [@puneet\_makhija](https://discuss.elastic.co/u/puneet_makhija)\
**Post date:** [March 4, 2022, 2:43pm UTC](https://discuss.elastic.co/t/deserialization-of-results-in-the-output-section/298634/7 "2022-03-04T14:43:22Z")

</div>

hi, anybody here who can help me in this, my build is stucked

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [April 1, 2022, 2:43pm UTC](https://discuss.elastic.co/t/deserialization-of-results-in-the-output-section/298634/8 "2022-04-01T14:43:45Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.