How to extract doc from avro data and add it to dataframe

Looking at: https://github.com/apache/spark/blob/master/connector/avro/src/main/scala/org/apache/spark/sql/avro/SchemaConverters.scala#L58

It seems like Spark's implementation of the schema converter does not call the getDoc function in the org.apache.avro.Schema object or the doc function of the org.apache.avro.Schema.Field objects it iterates through

It only reads the type metadata and tries to convert it to Spark's internal type objects (StructType, StringType, LongType, etc...)

The only 2 solutions I can think of are:

1. reading one or some of the files yourself and parsing the schema
2. Save the schema in a seperate file which should be easy enough to parse since its plain JSON

If you would like to parse the schema yourself and manually add metadata to spark, I would suggest flatdict package:

from flatdict import FlatterDict

flat_schema = FlatterDict(schema)  # schema as python dict

names = {k.replace(':name', ''): flat_schema[k] for k in flat_schema if k.endswith(':name')}
docs = {k.replace(':doc', ''): flat_schema[k] for k in flat_schema if k.endswith(':doc')}

# keep only keys which are present in both names and docs
keys_with_doc = set(names.keys()) & set(docs.keys())

full_name = lambda key: '_'.join(
    names[k] for k in sorted(names, key=len) if key.startswith(k) and k.split(':')[-2] == 'fields'
)
name_doc_map = {full_name(k): docs[k] for k in keys_with_doc}

A typical set of keys in flat_schema.keys() is:

'fields:1:type:1:fields:0:type:1:fields:0:type:1',
'fields:1:type:1:fields:0:type:1:fields:0:name',
'fields:1:type:1:fields:0:type:1:fields:0:default',
'fields:1:type:1:fields:0:type:1:fields:0:doc',

These strings can now be manipulated:

1. extract only the ones ending with "name" and "doc" (ignore "default", etc.)
2. get set intersection to remove the ones that do not have both fields present
3. get a list of all field names from higher levels of hierarchy: fields:1:type:1:fields is one of parents of fields:1:type:1:fields:0:type:1:fields (the condition is that they have the same start and they end with "fields")

Code from @bzu

from flatdict import FlatterDict

flat_schema = FlatterDict(schema)  # schema as python dict

names = {k.replace(':name', ''): flat_schema[k] for k in flat_schema if k.endswith(':name')}
docs = {k.replace(':doc', ''): flat_schema[k] for k in flat_schema if k.endswith(':doc')}

keys_with_doc = set(names.keys()) & set(docs.keys())

full_name = lambda key: '_'.join(
    names[k] for k in sorted(names, key=len) if key.startswith(k) and k.split(':')[-2] == 'fields'
)
name_doc_map = {full_name(k): docs[k] for k in keys_with_doc}

Explanation

FlatterDict

E.g.:

value = flatdict.FlatterDict({'list': ['a', 'b', 'c']})

will be the same as:

value == {'list:0': 'a', 'list:1': 'b', 'list:2': 'c'}

So basically your nested list basically will look like this:

{"fields0": ..., "fields1": ...}

More information about FlatterDict here

set(names.keys()) & set(docs.keys())

The names.keys() and docs.keys get all the keys from the dictionary, and currently there are multiple values, so we have to put them into a set to be grouped into one variable. The line then uses & to perform a bitwise operator on the two sets, essentially finding all duplicates in both sets and putting them into keys_with_doc. E.g.

a={1:"a", 3:"c", 4:"d"}
b={1:"a", 2:"b"}
name=set(a.keys()) & set(b.keys())
name

>>> {1}

Keys are the ones on the left:

dict = {1:"a",2:"b"}

full_name

This lambda function is tricky. It joins a bunch of stuff with "_". This "bunch of stuff" is the dictionary of names, which a loop will run through and extract names[k] from the sorted names dictionary using key=len and an if statement. Basically by using this lambda in name_doc_map, you are creating a new dictionary that resorts and conditions from the original set keys_with_doc along with docs.