我正试图通过以下siddhi程序从Kafka那里获取数据:
@source(type='kafka',
topic.list='matomo.matomo.matomo_site',
partition.no.list='0',
threading.option='single.thread',
group.id='consumer-group',
is.binary.message='true',
bootstrap.servers='kafka:9092',
@map(type='avro', schema.def = """{"type":"record","name":"Envelope","namespace":"matomo.matomo.matomo_site","fields":[{"name":"before","type":["null",{"type":"record","name":"Value","fields":[{"name":"idsite","type":"long"},{"name":"name","type":"string"},{"name":"main_url","type":"string"},{"name":"ts_created","type":["null",{"type":"string","connect.version":1,"connect.name":"io.debezium.time.ZonedTimestamp"}],"default":null},{"name":"ecommerce","type":[{"type":"int","connect.default":0,"connect.type":"int16"},"null"],"default":0},{"name":"sitesearch","type":[{"type":"int","connect.default":1,"connect.type":"int16"},"null"],"default":1},{"name":"sitesearch_keyword_parameters","type":"string"},{"name":"sitesearch_category_parameters","type":"string"},{"name":"timezone","type":"string"},{"name":"currency","type":"string"},{"name":"exclude_unknown_urls","type":[{"type":"int","connect.default":0,"connect.type":"int16"},"null"],"default":0},{"name":"excluded_ips","type":"string"},{"name":"excluded_parameters","type":"string"},{"name":"excluded_user_agents","type":"string"},{"name":"group","type":"string"},{"name":"type","type":"string"},{"name":"keep_url_fragment","type":{"type":"int","connect.default":0,"connect.type":"int16"},"default":0},{"name":"creator_login","type":["null","string"],"default":null}],"connect.name":"matomo.matomo.matomo_site.Value"}],"default":null},{"name":"after","type":["null","Value"],"default":null},{"name":"source","type":{"type":"record","name":"Source","namespace":"io.debezium.connector.mysql","fields":[{"name":"version","type":["null","string"],"default":null},{"name":"name","type":"string"},{"name":"server_id","type":"long"},{"name":"ts_sec","type":"long"},{"name":"gtid","type":["null","string"],"default":null},{"name":"file","type":"string"},{"name":"pos","type":"long"},{"name":"row","type":"int"},{"name":"snapshot","type":[{"type":"boolean","connect.default":false},"null"],"default":false},{"name":"thread","type":["null","long"],"default":null},{"name":"db","type":["null","string"],"default":null},{"name":"table","type":["null","string"],"default":null},{"name":"query","type":["null","string"],"default":null}],"connect.name":"io.debezium.connector.mysql.Source"}},{"name":"op","type":"string"},{"name":"ts_ms","type":["null","long"],"default":null}],"connect.name":"matomo.matomo.matomo_site.Envelope"}""",@attributes(idsite="idsite")))
define stream SweetProductionStream(idsite long);
@sink(type='log')
define stream LowProductionAlertStream(idsite long);
@info(name='outputQuery')
from SweetProductionStream
select *
insert into LowProductionAlertStream;
我收到一条错误消息:转换架构的avro消息时出错:。。。去西德的活动。8因此取消了该事件(编码的)
[2020-05-20_12-29-17_360] ERROR {io.siddhi.extension.map.avro.sourcemapper.AvroSourceMapper} - Error when converting avro message of schema: {"type":"record","name":"Envelope","namespace":"matomo.matomo.matomo_site","fields":[{"name":"before","type":["null",{"type":"record","name":"Value","fields":[{"name":"idsite","type":"long"},{"name":"name","type":"string"},{"name":"main_url","type":"string"},{"name":"ts_created","type":["null",{"type":"string","connect.version":1,"connect.name":"io.debezium.time.ZonedTimestamp"}],"default":null},{"name":"ecommerce","type":[{"type":"int","connect.default":0,"connect.type":"int16"},"null"],"default":0},{"name":"sitesearch","type":[{"type":"int","connect.default":1,"connect.type":"int16"},"null"],"default":1},{"name":"sitesearch_keyword_parameters","type":"string"},{"name":"sitesearch_category_parameters","type":"string"},{"name":"timezone","type":"string"},{"name":"currency","type":"string"},{"name":"exclude_unknown_urls","type":[{"type":"int","connect.default":0,"connect.type":"int16"},"null"],"default":0},{"name":"excluded_ips","type":"string"},{"name":"excluded_parameters","type":"string"},{"name":"excluded_user_agents","type":"string"},{"name":"group","type":"string"},{"name":"type","type":"string"},{"name":"keep_url_fragment","type":{"type":"int","connect.default":0,"connect.type":"int16"},"default":0},{"name":"creator_login","type":["null","string"],"default":null}],"connect.name":"matomo.matomo.matomo_site.Value"}],"default":null},{"name":"after","type":["null","Value"],"default":null},{"name":"source","type":{"type":"record","name":"Source","namespace":"io.debezium.connector.mysql","fields":[{"name":"version","type":["null","string"],"default":null},{"name":"name","type":"string"},{"name":"server_id","type":"long"},{"name":"ts_sec","type":"long"},{"name":"gtid","type":["null","string"],"default":null},{"name":"file","type":"string"},{"name":"pos","type":"long"},{"name":"row","type":"int"},{"name":"snapshot","type":[{"type":"boolean","connect.default":false},"null"],"default":false},{"name":"thread","type":["null","long"],"default":null},{"name":"db","type":["null","string"],"default":null},{"name":"table","type":["null","string"],"default":null},{"name":"query","type":["null","string"],"default":null}],"connect.name":"io.debezium.connector.mysql.Source"}},{"name":"op","type":"string"},{"name":"ts_ms","type":["null","long"],"default":null}],"connect.name":"matomo.matomo.matomo_site.Envelope"} to siddhi event. 8. Hence dropping the event. (Encoded)
请帮助我找出错误的原因,以及如何获得如此复杂方案的siddhi事件属性。
暂无答案!
目前还没有任何答案,快来回答吧!