Python Avro - Kafka Producer -


problem in short:

python avro kafka producer. can implement complex type avro format produce kfk message (just header avro type). but, when use header part of "rawmessage" avro, don't see way create message , publish kfk topic. help?

avro:

rawmessage - parent

{"namespace": "com.example",   "type": "record",   "name": "rawmessage",   "fields": [     {       "name": "header",       "type": "header"     },     {       "name": "payload",       "type": "string"     }   ] }

header -- child

{"namespace": "com.example",   "type": "record",   "name": "header",   "fields": [     {       "name": "channeltype",       "type": ["null", "string"] , "default" : null     },     {       "name": "flowtype",       "type": ["null", "string"] , "default" : null     },     {       "name": "type",       "type": "string" , "default" : "unassigned"     },     {       "name": "alerttype",       "type": ["null", "string"] , "default" : null     },     {       "name": "id",       "type": "string" , "default" : "unassigned"     },     {       "name": "version",       "type": "string" , "default" : "unassigned"     },     {       "name": "apikey",       "type": ["null", "string"] , "default" : null     },     {       "name": "sourcefile",       "type": ["null", "string"] , "default" : null     },     {       "name": "createdtimestamp",       "type": ["null", "string"] , "default" : null     },     {       "name": "enrichedheaders",       "type": ["null", {"type": "map", "values": "string"}]     }   ] }

python code::

from kafka import kafkaconsumer, kafkaproducer import avro.schema import io, random avro.datafile import datafilereader, datafilewriter avro.io import datumreader, datumwriter   # send messages synchronously bootstrap_servers_pt_list = 'localhost:9092' producer = kafkaproducer(bootstrap_servers=bootstrap_servers_pt_list)   # kafka topic topic = 'test.all.avro'    # path rawmessage.avsc avro schema schema_path = "/<path>/header.avsc" schema = avro.schema.parse(open(schema_path).read())  header = {     "channeltype": "channeltype",     "flowtype": "flowtype",     "type": "type",     "alerttype": "alerttype",     "id": "id",     "version": "version",     "apikey": "apikey",     "sourcefile": "aws_nrch_part_00000.dat",     "createdtimestamp": "20170504100126",     "enrichedheaders": {"filename": "20170504100126_eml_to_aws_nrch_part_00000.dat", "timestamp": "20170504100126"} }  # path rawmessage.avsc avro schema schema_path = "rawmessage.avsc" schema = avro.schema.parse(open(schema_path).read())   # path rawmessage.avsc avro schema schema_path = "/users/repo/rawmessage.avsc" schema_raw = avro.schema.parse(open(schema_path).read())   rawmessage = {     "header": header,     "payload": "payload" }   in xrange(10):     writer = avro.io.datumwriter(schema)     bytes_writer = io.bytesio()     encoder = avro.io.binaryencoder(bytes_writer)     writer.write(header, encoder)     raw_bytes = bytes_writer.getvalue()     producer.send(topic, raw_bytes)

question::

how use "header" part of "rawmessage" avro?? how make code work ?? possible in python avro have parent , child avro formats.. (i know can in java)


Comments

Post a Comment

Popular posts from this blog

javascript - Create a stacked percentage column -

i have collection of data : var data = [ {"p301a":"10","p301b":"7","p301c":"7","p301d":"3","p301e":"8","p301f":"8","p301g":"8","p301h":"8","p301i":"8","p301j":"8","p301k":"8","p301l":"8","p301m":"8","p301n":"8","p301o":"8","age":"31-40 years","profesion":"2","position":"2"}, {"p301a":"5","p301b":"4","p301c":"4","p301d":"4","p301e":"4","p301f":"4","p301g":"4","p301h":"4","p301i":"4","p301j":"4","p301k":"4...
Read more

Optimising Firebase database by automatically overwriting data -

in firebase app have following structure: posts latest all users can write posts > latest . want limit entries 20 posts save space in database. , posts beyond 20 unnecessary never shown on homepage. how can limit posts 20, when user writes posts > latest posts @ end drop off (be deleted) automatically? this sample should you. you need : const max_log_count = 20; exports.removeold = functions.database.ref('/posts/latest/{postid}').oncreate(event => { const parentref = event.data.ref.parent; return parentref.once('value').then(snapshot => { if (snapshot.numchildren() >= max_log_count) { let childcount = 0; const updates = {}; snapshot.foreach(function(child) { if (++childcount <= snapshot.numchildren() - max_log_count) { updates[child.key] = null; } }); // update parent. removes children. ...
Read more

javascript - Angular UI-Grid customTemplate directive causing rows to load slowly/? -

Image
each grid cell can have more 1 customtemplate loaded. have directive holds watcher on model_col_field listen cell updates on scroll when dom rows updated , recompile celltemplate . without watcher , $compile rows repeat on scroll, showing same set of rows. is there more angular way of approaching or way improve performance? relevant controller code: create gridoptions.data & gridoptions.columndefs : var customtemp = '<div ng-repeat="opt in grid.appscope.colvalueoptionsmodel" ng-init="tempstring=model_col_field.clazz+opt.key;"><comparison-directive dbo="model_col_field" attobj="opt" fff="grid.appscope.getattributetemplate(tempstring)" /></div>'; $scope.gridoptions.columndefs = [ { name:'yindex',pinnedleft:true,celltemplate: customtemp, displayname:'index'}, { name:'col1',celltemplate: customtemp}, { name:'col2',celltemplate: c...
Read more