# Create stream from Avro topics generated from DataGen

**URL:** <https://forum.confluent.io/t/create-stream-from-avro-topics-generated-from-datagen/10434>\
**Category:** Kafka Connect\
**Created:** [21 March 2024 19:23 UTC](https://forum.confluent.io/t/create-stream-from-avro-topics-generated-from-datagen/10434 "2024-03-21T19:23:11Z")\
**Posts on this page:** 7\
**Page:** 1

<div class="post-metadata">

**Author:** ![shjoon](https://avatars.discourse-cdn.com/v4/letter/s/87869e/32.png) [@shjoon](https://forum.confluent.io/u/shjoon)\
**Post date:** [21 March 2024 19:23 UTC](https://forum.confluent.io/t/create-stream-from-avro-topics-generated-from-datagen/10434/1 "2024-03-21T19:23:11Z")

</div>

in confluent cloud, i am using DataGen connector . I am making DataGen connector to generate message in topic with Avro format.

the datagen has created a topic already with Avro format.

I want to make a stream from it in Ksqldb. I have executed command like this  
The topic name that I put in the WITH, it already exist from DataGen,

it creates the stream but when i query the stream with emit changes, it keeps running but no message is returned.

is this because of Avro serialization? if yes, is it possible to create stream from AVRO topics ? what should i change in my create stream command ?

```
		CREATE STREAM TcProductVariant_stream_Avro (

```

\_comment string ,  
data STRUCT\<  
productInstanceUuid string ,  
productInstanceId string,  
productInstanceDescription string ,  
productInstanceRevision string ,  
productVariants ARRAY\<STRUCT\<operation string , TC\_variant\_id string , tc\_variant\_name string, variantOptions string

> >

)  
WITH (KAFKA\_TOPIC = ‘TcProductVariants\_Avro’,  
VALUE\_FORMAT = ‘avro’,  
TIMESTAMP\_FORMAT = ‘yyyy-MM-dd HH:mm:ss’,  
PARTITIONS = 6);

---

<div class="post-metadata">

**Author:** ![dtroiano](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/dtroiano/32/1961_2.png) [@dtroiano](https://forum.confluent.io/u/dtroiano)\
**Post date:** [21 March 2024 21:46 UTC](https://forum.confluent.io/t/create-stream-from-avro-topics-generated-from-datagen/10434/2 "2024-03-21T21:46:20Z")

</div>

Yes, you can create a stream backed by a topic populated by a Datagen connector that generates Avro-formatted data. KsqlDB will pick up the schema so you don’t even need to specify columns when creating the stream. I.e., just this should work:

```auto
CREATE STREAM TcProductVariant_stream_Avro
WITH (KAFKA_TOPIC='TcProductVariants_Avro', VALUE_FORMAT='AVRO');

```

[Here](https://developer.confluent.io/tutorials/generate-streams-of-test-data/ksql.html) is a tutorial that does this with one of Datagen’s built-in data sets.

I’m not sure why you aren’t seeing results when you query the stream. Possibly the connector had stopped and you queried with the default `auto.offset.reset` value of `latest`? As long as there’s some data in the topic, then you should see results if you set `auto.offset.reset` to `earliest` and query the stream:

```auto
SET 'auto.offset.reset'='earliest';
SELECT * FROM TcProductVariant_stream_Avro EMIT CHANGES;

```

---

<div class="post-metadata">

**Author:** ![shjoon](https://avatars.discourse-cdn.com/v4/letter/s/87869e/32.png) [@shjoon](https://forum.confluent.io/u/shjoon)\
**Post date:** [22 March 2024 16:14 UTC](https://forum.confluent.io/t/create-stream-from-avro-topics-generated-from-datagen/10434/3 "2024-03-22T16:14:49Z")

</div>

What can be the reason this statement is failing, it is saying

Exception while preparing statement: PRODUCTINSTANCE\_VARIANT\_OPTIONS does not exist.

I have that topic that is generated from data gen.

CREATE or replace STREAM TC\_VARIANT\_OPTIONS\_STREAM\_V2 WITH  
(  
KAFKA\_TOPIC=‘TC\_VARIANT\_OPTIONS’,  
PARTITIONS=3, REPLICAS=3,  
RETENTION\_MS=604800000, TIMESTAMP\_FORMAT=‘yyyy-MM-dd HH:mm:ss’, VALUE\_FORMAT=‘JSON’)  
AS select  
rowtime as source\_row\_time,  
format\_timestamp(FROM\_UNIXTIME(rowtime),‘yyyy-MM-dd HH:mm:ss.SSS’, ‘UTC’) as message\_time,  
explode(VARIANTOPTIONS) as original ,  
json\_keys(explode(VARIANTOPTIONS)) as json\_keys ,  
json\_records(explode(VARIANTOPTIONS)) as json\_records,  
map\_keys(json\_records(explode(VARIANTOPTIONS))) as map\_keys  
from PRODUCTINSTANCE\_VARIANT\_OPTIONS ;

while  
if i change the FROM to refer to a stream called PRODUCTINSTANCE\_VARIANT\_OPTIONS\_STREAM that is what i created a stream from that topic, it will work.

what is the reason of failing pointing to topic ?

---

<div class="post-metadata">

**Author:** ![dtroiano](https://sea1.discourse-cdn.com/flex019/user_avatar/forum.confluent.io/dtroiano/32/1961_2.png) [@dtroiano](https://forum.confluent.io/u/dtroiano)\
**Post date:** [22 March 2024 17:52 UTC](https://forum.confluent.io/t/create-stream-from-avro-topics-generated-from-datagen/10434/4 "2024-03-22T17:52:31Z")

</div>

`PRODUCTINSTANCE_VARIANT_OPTIONS ` is only a topic name? You can’t directly select from a topic. You have to define a stream that wraps the topic and select from the stream.

---

<div class="post-metadata">

**Author:** ![shjoon](https://avatars.discourse-cdn.com/v4/letter/s/87869e/32.png) [@shjoon](https://forum.confluent.io/u/shjoon)\
**Post date:** [22 March 2024 20:14 UTC](https://forum.confluent.io/t/create-stream-from-avro-topics-generated-from-datagen/10434/5 "2024-03-22T20:14:30Z")

</div>

thanks coming back to original problem.

I stilll can’t create a avro stream from avro topic.

I attached example of avro message in topic

i am creating this stream

CREATE STREAM xyz (  
\_comment string ,  
data STRUCT\<  
productInstanceUuid string ,  
productInstanceId string,  
productInstanceDescription string ,  
productInstanceRevision string ,  
productVariants ARRAY\<STRUCT\<operation string , TC\_variant\_id string , tc\_variant\_name string, variantOptions ARRAY\< MAP\<string,string\>\>

> >

)  
WITH (KAFKA\_TOPIC = ‘TcProductVariants\_Avro’,  
VALUE\_FORMAT = ‘avro’,  
TIMESTAMP\_FORMAT = ‘yyyy-MM-dd HH:mm:ss’,  
PARTITIONS = 6);

I am getting this error in ksql processing log

“DESERIALIZATIONERRORCAUSE”: [  
“Cannot deserialize type array as type string for path: -\>DATA-\>PRODUCTVARIANTS-\>ARRAY-\>VARIANTOPTIONS”  
],  
“DESERIALIZATIONERRORMESSAGE”: “Error deserializing message from topic: TcProductVariants\_Avro”,

example of my message in topic is this

{  
“\_comment”: “s)CPt}\n7I\u0002\>q\u001b\u001b”,  
“data”: {  
“productInstanceUuid”: “8d0d1ce3-53ac-4704-9b6e-6753e9b1c4ae”,  
“productInstanceId”: “P999888777”,  
“productInstanceDescription”: “City Hybrid”,  
“productInstanceRevision”: “A”,  
“productVariants”: [  
{  
“operation”: “Add”,  
“Tc\_variant\_id”: “c4bf6e3d-bf7d-4950-93da-d59d21c29ec9”,  
“Tc\_variant\_name”: “red AND small”,  
“VariantOptions”: [  
{  
“color”: “red”,  
“quantity”: “small”  
}  
]  
}  
}  
]  
}  
}

---

<div class="post-metadata">

**Author:** ![shjoon](https://avatars.discourse-cdn.com/v4/letter/s/87869e/32.png) [@shjoon](https://forum.confluent.io/u/shjoon)\
**Post date:** [25 March 2024 18:03 UTC](https://forum.confluent.io/t/create-stream-from-avro-topics-generated-from-datagen/10434/6 "2024-03-25T18:03:21Z")

</div>

is there a hope in here to get a hand helping ?

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/flex019/uploads/confluentcommunity/original/1X/c49438c90c9df282e9996fdf6971be890c71b65a.svg) [@system](https://forum.confluent.io/u/system)\
**Post date:** [24 April 2024 18:04 UTC](https://forum.confluent.io/t/create-stream-from-avro-topics-generated-from-datagen/10434/7 "2024-04-24T18:04:15Z")

</div>

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