I want to add avro schema for nested json which I am trying to read from kafka example json : { "payload":{ "before":null, "after":{ "id":1 }, "source":{ "version":"0.8.1.Final", "name":"dev", "server_id":0, "ts_sec":0, "gtid":null, "file":"mysql-bin-changelog.022566", "pos":154, "row":0, "snapshot":true, "thread":null, "db":"dev_datalake_poc", "table":"dummy", "query":null }, "op":"c", "ts_ms":1534239705758 } } Supposingly if I want to fetch payload.after.id and payload.op I have tried and it results in null values ("payload" :null) : { "type" : "record", "name" : "dummyschema", "fields" : [ { "name" : "payload", "type" : { "type" : "record", "name" : "payload", "fields" : [ { "name" : "before", "type" : [ "string", "null" ] }, { "name" : "after", "type" : { "type" : "record", "name" : "after", "fields" : [ { "name" : "id", "type" : "long" } ] } }, { "name" : "source", "type" : { "type" : "record", "name" : "source", "fields" : [ { "name" : "version", "type" : "string" }, { "name" : "name", "type" : "string" }, { "name" : "server_id", "type" : "long" }, { "name" : "ts_sec", "type" : "long" }, { "name" : "gtid", "type" : [ "string", "null" ] }, { "name" : "file", "type" : "string" }, { "name" : "pos", "type" : "long" }, { "name" : "row", "type" : "long" }, { "name" : "snapshot", "type" : "boolean" }, { "name" : "thread", "type" : [ "string", "null" ] }, { "name" : "db", "type" : "string" }, { "name" : "table", "type" : "string" }, { "name" : "query", "type" : [ "string", "null" ] } ] } }, { "name" : "op", "type" : "string" }, { "name" : "ts_ms", "type" : "long" } ] } } ] } ... View more