在实际工作中，会将avro文件读到一个类中。下面就看看代码是如何实现的（这里看的是avro-1.7.7版本）。
读取整个avro文件的类是DataFileReader，用于读取data的类是GenericDatumReader（data指的是用户的数据），有关avro文件结构可参见avro源码阅读-写avro文件中的图。
我们从AvroPairInputFormat切入看下DataFileReader和SpecificDatumReader的使用。
如果你看过了avro源码阅读-写avro文件，那你会觉得很熟悉。
在读avro文件时，有DataFileReader和AvroPairInputFormat。
在写avro文件时，有DataFileWrite和AvroPairOutputFormat。
AvroInputFormat
- AvroInputFormat.getRecordReader()返回的是AvroRecordReader；
- AvroRecordReader.next()读取下一条记录；
- AvroRecordReader.next()又调用了DataFileReader.next()；
- DataFileReader.next()调用了GenericDatumReader.read()。
其中， - DataFileReader的next()和reader继承自DataFileStream；
- GenericDatumReader.read(reuse, decoder)参数中的decoder是BinaryDecoder。
类的继承及引用关系图如下：
对于DataFileReader读取整个avro文件的代码这里不再详述，这里主要关心GenericDatumReader是如何读取data的。
GenericDatumReader
GenericDatumReader.read()的构造了一个ResolvingDecoder。ResolvingDecoder维护了一个解码器Decoder和解析器SkipParser。SkipParser维护了一个语法堆栈，用于指导下一步该如何处理（如是读数据还是跳过，若读数据是什么类型的数据）。Decoder实际上是BinaryDecoder，在需要读取数据时，由它从avro文件中读取一小块内容到内存中。ResolvingDecoder还引用了ResolvingGrammarGenerator，该类用于生成语法，语法由Symbol类表示。这几个类的关系图如下所示：
Symbol
SkipParser维护了一个语法堆栈（从Parser继承而来，实现上是一个数组），Symbol就是存储在堆栈中的元素。Symbol就像是写avro文件时用到的Schema。写avro文件时，是将一条数据按照schema写到文件中；而读avro文件时，则是按照Symbol将一条数据读取到内存中。SkipParser
SkipParser维护了一个Symbol的数组，也就是一个堆栈，每次弹出栈顶元素，然后执行相应动作：
-将栈顶元素扩展后再压栈；
-直接从avro文件中读取一小块数据；
-其他操作比如跳过。ResolvingGrammarGenerator
ResolvingGrammarGenerator用于生成压入堆栈中的第一个元素（根元素）。
代码里对这几个类的注释：
Symbol：Symbol is the base of all symbols (terminals and non-terminals) of the grammar.
SkipParser：A parser that capable of skipping as well read and write. This class is used by decoders who (unlink encoders) are required to implement methods to skip.
ResolvingGrammarGenerator：The class that generates validating grammar.
Symbol
和Schema一样，Symbol是个抽象类，它有很多具体的实现。这里关注三种实现：
- Terminal： 一个基本类型如int、long等，就对应一个terminal。当弹出的栈顶元素为terminal时，就表示应该从数据中读取某个类型对应大小的bytes了。
-
Sequence：比如说一个类，就对应一个Sequence。
Symbol本身又维护了一个Symbol数组production，production表示该Symbol的产物。
Symbol本身包含一个Symbol数组，就像类包含的成员变量也可以是其他类一样。
他们的对应关系可描述为下图：
Sequence和Class的对应关系
Terminal的production是空的，而Sequence的production包含了其产物。
- Root：根Symbol，就是第一个压入语法堆栈的元素。
GenericDatumReader.read()代码如下：
<pre>
public D read(D reuse, Decoder in) throws IOException {
ResolvingDecoder resolver = getResolver(actual, expected); //ResolvingDecoder的初始化
resolver.configure(in); //将ResolvingDecoder 的解码器配置为in（即BInaryDecoder）
D result = (D) read(reuse, expected, resolver); //读取一条记录到reuse中
resolver.drain();
return result;
}
</pre>
下面分别介绍：
- ResolvingDecoder的初始化；
- 读取一条记录到reuse中
ResolvingDecoder的初始化
ResolvingDecoder初始化的时候，new了一个SkipParser，生成Root Symbol并将其压入了SkipParser维护的堆栈中。初始化中复杂之处在于Root Symbol的生成，生成Root Symbol的代码如下：
<pre>
return Symbol.root(generate(writer, reader, new HashMap<LitS, Symbol>()));
</pre>
这里有两步调用：
- generate(writer, reader, map);
- Symbol.root(symbol)
generate(writer, reader, map)
writer指的是avro文件中记录的schema，reader指的是类的schema。
当writer和reader的类型不同时，generate()会推测可能是UNION导致的，或者进行一些类型转换，否则会抛出异常；
-
当writer和reader的类型相同时，代码如下：
<pre>
public Symbol generate(Schema writer, Schema reader,
Map<LitS, Symbol> seen) throws IOException {
final Schema.Type writerType = writer.getType();
final Schema.Type readerType = reader.getType();if (writerType == readerType) {
switch (writerType) {
case NULL:
return Symbol.NULL;
case BOOLEAN:
return Symbol.BOOLEAN;
case INT:
return Symbol.INT;
case LONG:
return Symbol.LONG;
...
case RECORD:
return resolveRecords(writer, reader, seen);
...
}
</pre>
即使类型相同，writer和reader的schema也可能有所差别，比如用新的类（可能增加了一些字段，或是字段顺序有所调整）来读取历史文件。resolveRecords()会综合writerSchema和readerSchema中的字段，生成并返回一个Symbol。主要的情况有下面两种：
-wirterSchema中有，但是readerSchema中没有的字段，将会被跳过；
-readerSchema中有，但是writerSchema中没有的字段，必须有默认值，否则就会出错。
resolveRecords()的代码如下：
<pre>
private Symbol resolveRecords(Schema writer, Schema reader,
Map<LitS, Symbol> seen) throws IOException {
LitS wsc = new LitS2(writer, reader);
Symbol result = seen.get(wsc); //先从缓存中取（根据writer和reader构造symbol也是一件比较耗时的事情）
if (result == null) {
List<Field> wfields = writer.getFields();
List<Field> rfields = reader.getFields();
// First, compute reordering of reader fields, plus
// number elements in the result's production
Field[] reordered = new Field[rfields.size()];
int ridx = 0;
int count = 1 + wfields.size(); //count是Symbol的产物production的大小，也就是writerSchema展开后字段的个数（既然是Record类型，肯定是有产物的嘛）
//这里的"1"被fieldOrderAction占用的
//将在writerSchema和readerSchema中都存在的字段，按照writerSchema中的顺序，存储在reordered中
//因为最终是要从文件中依次将数据读出来，所以要按照writerSchema中的字段顺序
for (Field f : wfields) {
Field rdrField = reader.getField(f.name());
if (rdrField != null) {
reordered[ridx++] = rdrField;
}
}
//处理readerSchema中有，而writerSchema中没有的字段（如类新增了一个字段，这在历史数据中是没有的）
//此时，该字段必须有自己的默认值，否则就会返回错误信息
for (Field rf : rfields) {
String fname = rf.name();
if (writer.getField(fname) == null) {
if (rf.defaultValue() == null) {
result = Symbol.error("Found " + writer.getFullName()
+ ", expecting " + reader.getFullName()
+ ", missing required field " + fname);
seen.put(wsc, result);
return result;
} else {
reordered[ridx++] = rf;
count += 3;//使用默认值的字段需要占用3个位置
}
}
}
Symbol[] production = new Symbol[count]; //初始化production数组
production[--count] = Symbol.fieldOrderAction(reordered); //FieldOrderAction记录了production中其他各个字段的schema
/**
- We construct a symbol without filling the array. Please see
- {@link Symbol#production} for the reason.
*/
result = Symbol.seq(production); //生成了一个Sequence Symbol
seen.put(wsc, result); //缓存
/*
- For now every field in read-record with no default value
- must be in write-record.
- Write record may have additional fields, which will be
- skipped during read.
*/
// Handle all the writer's fields
for (Field wf : wfields) {
String fname = wf.name();
Field rf = reader.getField(fname);
if (rf == null) { //如果writerSchema中有的字段，readerSchema中没有（比如类的某些字段删掉了，但是历史数据中还存在），则标识为跳过
production[--count] =
Symbol.skipAction(generate(wf.schema(), wf.schema(), seen));
} else {
production[--count] =
generate(wf.schema(), rf.schema(), seen); //递归的调用generate()（产物production中可能也有一些复杂的结构）
}
}
// Add default values for fields missing from Writer
for (Field rf : rfields) {
String fname = rf.name();
Field wf = writer.getField(fname);
if (wf == null) { //处理readerSchema中有，而writerSchema中没有，但是有默认值的字段（上面提到过了），可以看到，这里需要使用三个位置
byte[] bb = getBinary(rf.schema(), rf.defaultValue());
production[--count] = Symbol.defaultStartAction(bb);
production[--count] = generate(rf.schema(), rf.schema(), seen);
production[--count] = Symbol.DEFAULT_END_ACTION;
}
}
}
return result;
}
</pre>
举个栗子，如writerSchema和readerSchema都是下面这样。
<pre>
{
"type": "record",
"name": "User",
"namespace": "test",
"fields": [
{
"name": "id",
"type": "int"
},
{
"name": "name",
"type": "string",
"default": null
},
{
"name": "dog",
"type": {
"name": "Dog",
"type": "record",
"fields": [
{
"name": "id",
"type": "int"
},
{
"name": "name",
"type": "string",
"default": null
}
]
}
}
]
}
</pre>
最终generate返回的Symbol结构如下（图最左边的sequenceSymbol0即是返回的结果。图中方框左侧的数字代表production数组的下标，注意，各个Symbol存储的顺序和schema中的顺序正好相反）。
Symbol.root(symbol)
root()会new一个RootSymbol，并将上面生成的sequenceSymbol0展开，作为该RootSymbol的production。这里最复杂的就在于将sequenceSymbol0展开的操作flatten()。这里关注两个flatten()。
-
Symbol类的静态方法
将In:Symbol[]平铺开为out:Symbol[]，大概就是下面这么个意思。
in2out
代码如下：
<pre>
static void flatten(Symbol[] in, int start,
Symbol[] out, int skip,
Map<Sequence, Sequence> map,
Map<Sequence, List<Fixup>> map2) {
for (int i = start, j = skip; i < in.length; i++) {
Symbol s = in[i].flatten(map, map2); //如果in[i]是一个Terminal，则直接返回其本身
//如果in[i]是一个Sequence，则递归的展开自身
//其他类型的Symbol在此不再关注
if (s instanceof Sequence) {
Symbol[] p = s.production;
List<Fixup> l = map2.get(s);
if (l == null) {
System.arraycopy(p, 0, out, j, p.length);
} else {
l.add(new Fixup(out, j));
}
j += p.length;
} else { //如果不是Sequence，直接将in赋值给out
out[j++] = s;
}
}
}
</pre> -
Sequence的私有方法
将sequence的production铺开。将上图"in2out"中的in定义为某个Sequence展开前prodcution，out定义为某个Sequence展开后的production，则上图即可表示这个方法的意思。
<pre>
@Override
public Sequence flatten(Map<Sequence, Sequence> map,
Map<Sequence, List<Fixup>> map2) {
Sequence result = map.get(this);
if (result == null) {
result = new Sequence(new Symbol[flattenedSize()]); //初始化展开后的Sequence（flattenedSize也是个递归调用，用于获取一个Sequence的production层层展开后的Symbol的个数）
map.put(this, result); //缓存展开结果
List<Fixup> l = new ArrayList<Fixup>();
map2.put(result, l);flatten(production, 0,
result.production, 0, map, map2);
for (Fixup f : l) {
System.arraycopy(result.production, 0, f.symbols, f.pos,
result.production.length);
}
map2.remove(result);
}
return result;
}
</pre>
最简单的情况下，假设sequenceSymbol0只有一堆terminalSymbol，那就只调用了Symbol类静态的flatten()，直接将in赋值给了out。
复杂一点的情况下，如sequenceSymbol0还包含了sequenceSymbol1，或是嵌套着更多类的引用。此时，两个flatten()会相互递归调用，但最终，总会有一个sequence只包含Terminal之类的Symbol，从而结束递归。此时，代码中的map可以用来缓存展开结果，比如ClassA中有两个ClassB成员的情况。
不过，还有个关键的地方没弄懂。据注释所说，map、map2以及Fixup可用来解决无限递归的。比如，设result为SequenceA扩展后的结果，在result的production填充之前，便存储在map中。下次因递归而再次遇到SequenceA时，则直接从map中取到result。但此时，result的production并未被填充，而是依赖后续的Fixup进行填充。
可能产生无限递归的情况我只想到一种，那就是两个类之间相互引用（即ClassA引用了ClassB，ClassB又引用了ClassA）。但是代码最终未能调试起来，因为在这种情况下，flattenedSize()没能规避无限递归，导致了StackOverflowError。
导致问题的代码如下。
类A
<pre>
public class A {
public B b;
}
</pre>
类B
<pre>
public class B {
public A a;
}
</pre>
主函数
<pre>
public static void main(String[] args) throws IOException {
Schema s = ReflectData.get().getSchema(A.class);
DecoderFactory.get().resolvingDecoder(
Schema.applyAliases(s, s), s, null);
}
</pre>
仍以上面的sequenceSymbol0为例，最终生成的Root Symbol及语法堆栈如下图所示。
读取一条记录到reuse中
此时，resolver已经初始化好了，并且语法堆栈已经有了一个Root Symbol，接下来就是基于语法堆栈读取数据。
read(reuse, expected, resolver)根据expected（也就是类的schema）的类型不同，调用不同的方法，代码如下。
<pre>
protected Object read(Object old, Schema expected,
ResolvingDecoder in) throws IOException {
switch (expected.getType()) {
case RECORD: return readRecord(old, expected, in);
case ENUM: return readEnum(expected, in);
....
}
}
</pre>
这里主要还是看readRecord()，代码如下。
<pre>
protected Object readRecord(Object old, Schema expected,
ResolvingDecoder in) throws IOException {
...
for (Field f : in.readFieldOrder()) { //获取record的fields
...
readField(r, f, oldDatum, in, state); //读取field，其中又递归的调用了read()
}
...
}
</pre>
其中，for循环内递归调用了上面的read()，不再细述。这里主要看一下代码是如何获取record的fields的。
readFieldOrder()就一行代码，调用了Parser.advance()。
<pre>return ((Symbol.FieldOrderAction) parser.advance(Symbol.FIELD_ACTION)).fields;</pre>
Parser维护了语法堆栈，advance()顾名思义，也即弹出栈顶，并据此进行一次处理。
advance()代码如下。
<pre>
public final Symbol advance(Symbol input) throws IOException {
for (; ;) {
Symbol top = stack[--pos]; //栈顶出栈
if (top == input) { //如top和input都是一个terminal，像readDouble()、readInt()这些方法，就会执行这个分支
return top; // A common case
}
Symbol.Kind k = top.kind;
if (k == Symbol.Kind.IMPLICIT_ACTION) { //一个隐式动作，这里的doAction()其实就是ResolvingDecoder.doAction()
Symbol result = symbolHandler.doAction(input, top);
if (result != null) {
return result;
}
} else if (k == Symbol.Kind.TERMINAL) { //如果input是一个terminal，但是栈顶却不是terminal，抛出异常
throw new AvroTypeException("Attempt to process a "
+ input + " when a "
+ top + " was expected.");
} else if (k == Symbol.Kind.REPEATER
&& input == ((Symbol.Repeater) top).end) {
return input;
} else {
pushProduction(top); //将栈顶的production压栈
}
}
}
</pre>
第一个栈顶是Root Symbol，所以执行的是最后一个分支：pushProduction()。pushProdcution将栈顶（刚刚出栈）的production压栈。如果堆栈大小不够，扩展堆栈。
<pre>
public final void pushProduction(Symbol sym) {
Symbol[] p = sym.production;
while (pos + p.length > stack.length) { //大小不够，则扩栈
expandStack();
}
System.arraycopy(p, 0, stack, pos, p.length);
pos += p.length;
}
</pre>
执行完pushProduction()后，堆栈成为了下面这个样子。
之后会继续执行for()循环，继续弹出栈顶。
此时的栈顶是一个FieldOrderAction1，FieldOrderAction属于IMPLICIT_ACTION。此时，将会执行symbolHandler.doAction()。
这里的symbolHandler其实就是ResolvingDecoder，从ResolvingDecoder.doAction()中可以看到哪些Symbol属于IMPLICT_ACTION。这里不再深入。
doAction()的开头就是对FieldOrderAction的处理，可以看到直接返回了top（也即返回了FieldOrderAction1)。
<pre>
public Symbol doAction(Symbol input, Symbol top) throws IOException {
if (top instanceof Symbol.FieldOrderAction) {
return input == Symbol.FIELD_ACTION ? top : null;
...
}
</pre>
这就将FieldOrderAction1取出来了。FieldOrderAction1中包含了intSchema、stringSchema和recordSchema，在遍历到recordSchema时，又会递归的调用readRecord()。
left
还有问题没搞清楚，先搁着吧。
RootSymbol的production为何提前铺开了？不铺开行么？
GenericDatumReader.read()返回的是GenericData，GenericData最终是如何映射到类上的？
