[Avro]Avro入门例子

1. Maven配置

pom.xml

1. <?xml version="1.0" encoding="UTF-8"?>
2. <project xmlns="http://maven.apache.org/POM/4.0.0"
3.         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4.         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5.    <parent>
6.        <artifactId>open-diary</artifactId>
7.        <groupId>com.sjf.open</groupId>
8.        <version>1.0-SNAPSHOT</version>
9.    </parent>
10.    <modelVersion>4.0.0</modelVersion>
12.    <artifactId>open.avro</artifactId>
14.    <dependencies>
15.        <dependency>
16.            <groupId>org.apache.avro</groupId>
17.            <artifactId>avro</artifactId>
18.        </dependency>
19.    </dependencies>
21.    <build>
22.        <plugins>
23.            <plugin>
24.                <groupId>org.apache.avro</groupId>
25.                <artifactId>avro-maven-plugin</artifactId>
26.                <version>1.8.1</version>
27.                <executions>
28.                    <execution>
29.                        <phase>generate-sources</phase>
30.                        <goals>
31.                            <goal>schema</goal>
32.                        </goals>
33.                        <configuration>
34.                            <sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory>
35.                            <outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
36.                        </configuration>
37.                    </execution>
38.                </executions>
39.            </plugin>
41.            <plugin>
42.                <groupId>org.apache.maven.plugins</groupId>
43.                <artifactId>maven-compiler-plugin</artifactId>
44.                <configuration>
45.                    <source>1.6</source>
46.                    <target>1.6</target>
47.                </configuration>
48.            </plugin>
49.        </plugins>
50.    </build>
52. </project>

2. 定义模式(schema)

Avro的数据格式(schema)采用JSON定义，shcema由原始类型（null，boolean，int，long，float，double，bytes，string）和复杂类型（record，enum，array，map，union，fixed）组成，schema文件通常以.avsc结尾，代表avro schema。例如我们定义一个代表User的schema，文件名为user.avsc，放在src/main/avro目录下，这个必须与pom.xml文件配置的一致：

1. <sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory>

下面来定义模式：

1. {
2.  "namespace":"com.sjf.open",
3.  "type":"record",
4.  "name":"User",
5.  "fields":[
6.    {"name":"name","type":"string"},
7.    {"name":"age","type":["int","null"]},
8.    {"name":"favorite","type":["string","null"]}
9.  ]
10. }

上述定义中，namespace和name共同组成schema 的全称: com.sjf.open.User，表示将会在com.sjf.open包下创建User对象。type指定类型为复杂类型record，fileds数组定义record的字段。字段中的type指定为数组，代表该字段可以是int或者null类型。每一种数据类型可以定义哪一些参数，由Avro规范定义，也就是规范了定义Scheme中JSON需要遵循的JSON Schema。

一个模式文件只能包含一个模式（schema）。

3. 编译模式

Avro允许我们基于我们之前定义的模式来创建相应的类。一旦我们已经定义了相关的类，则程序中就不需要直接使用模式。可以使用avro-tools jar来生成代码如下:

1. java -jar /path/to/avro-tools-1.8.1.jar compile schema <schema file> <destination>

这将会在所提供的目标文件夹（<destination>）下基于之前配置模式的namespace字段，生成一个包，并在包中生成相应的类文件。在我们例子中，

1. java -jar /home/xiaosi/lib/avro-tools-1.8.1.jar compile schema user.avsc .

这样，就在当前根目录下生成com/sjf/open/User.java类，其中com/sjf/open就是我们在模式中的namespace。

如果你直接使用Avro Maven plugin，那么就不需要手动编译模式，因为Avro Maven plugin会自动给你编译好。

下面为Avro Maven plugin配置：

1.            <plugin>
2.                <groupId>org.apache.avro</groupId>
3.                <artifactId>avro-maven-plugin</artifactId>
4.                <version>1.8.1</version>
5.                <executions>
6.                    <execution>
7.                        <phase>generate-sources</phase>
8.                        <goals>
9.                            <goal>schema</goal>
10.                        </goals>
11.                        <configuration>
12.                            <sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory>
13.                            <outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
14.                        </configuration>
15.                    </execution>
16.                </executions>
17.            </plugin>
19.            <plugin>
20.                <groupId>org.apache.maven.plugins</groupId>
21.                <artifactId>maven-compiler-plugin</artifactId>
22.                <configuration>
23.                    <source>1.6</source>
24.                    <target>1.6</target>
25.                </configuration>
26.            </plugin>

现在通过编译模式已经生成一个User.java类：

1. /**
2. * Autogenerated by Avro
3. *
4. * DO NOT EDIT DIRECTLY
5. */
6. package com.sjf.open;
8. import org.apache.avro.specific.SpecificData;
10. @SuppressWarnings("all")
11. @org.apache.avro.specific.AvroGenerated
12. public class User extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
13.  private static final long serialVersionUID = 3024021648306482970L;
14.  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"User\",\"namespace\":\"com.sjf.open\",\"fields\":[{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"age\",\"type\":[\"int\",\"null\"]},{\"name\":\"favorite\",\"type\":[\"string\",\"null\"]}]}");
15.  public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
16.  @Deprecated public java.lang.CharSequence name;
17.  @Deprecated public java.lang.Integer age;
18.  @Deprecated public java.lang.CharSequence favorite;
20.  /**
21.   * Default constructor.  Note that this does not initialize fields
22.   * to their default values from the schema.  If that is desired then
23.   * one should use <code>newBuilder()</code>.
24.   */
25.  public User() {}
27.  /**
28.   * All-args constructor.
29.   * @param name The new value for name
30.   * @param age The new value for age
31.   * @param favorite The new value for favorite
32.   */
33.  public User(java.lang.CharSequence name, java.lang.Integer age, java.lang.CharSequence favorite) {
34.    this.name = name;
35.    this.age = age;
36.    this.favorite = favorite;
37.  }
39.  public org.apache.avro.Schema getSchema() { return SCHEMA$; }
40.  // Used by DatumWriter.  Applications should not call.
41.  public java.lang.Object get(int field$) {
42.    switch (field$) {
43.    case 0: return name;
44.    case 1: return age;
45.    case 2: return favorite;
46.    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
47.    }
48.  }
50.  // Used by DatumReader.  Applications should not call.
51.  @SuppressWarnings(value="unchecked")
52.  public void put(int field$, java.lang.Object value$) {
53.    switch (field$) {
54.    case 0: name = (java.lang.CharSequence)value$; break;
55.    case 1: age = (java.lang.Integer)value$; break;
56.    case 2: favorite = (java.lang.CharSequence)value$; break;
57.    default: throw new org.apache.avro.AvroRuntimeException("Bad index");
58.    }
59.  }
61.  /**
62.   * Gets the value of the 'name' field.
63.   * @return The value of the 'name' field.
64.   */
65.  public java.lang.CharSequence getName() {
66.    return name;
67.  }
69.  /**
70.   * Sets the value of the 'name' field.
71.   * @param value the value to set.
72.   */
73.  public void setName(java.lang.CharSequence value) {
74.    this.name = value;
75.  }
77.  /**
78.   * Gets the value of the 'age' field.
79.   * @return The value of the 'age' field.
80.   */
81.  public java.lang.Integer getAge() {
82.    return age;
83.  }
85.  /**
86.   * Sets the value of the 'age' field.
87.   * @param value the value to set.
88.   */
89.  public void setAge(java.lang.Integer value) {
90.    this.age = value;
91.  }
93.  /**
94.   * Gets the value of the 'favorite' field.
95.   * @return The value of the 'favorite' field.
96.   */
97.  public java.lang.CharSequence getFavorite() {
98.    return favorite;
99.  }
101.  /**
102.   * Sets the value of the 'favorite' field.
103.   * @param value the value to set.
104.   */
105.  public void setFavorite(java.lang.CharSequence value) {
106.    this.favorite = value;
107.  }
109.  /**
110.   * Creates a new User RecordBuilder.
111.   * @return A new User RecordBuilder
112.   */
113.  public static com.sjf.open.User.Builder newBuilder() {
114.    return new com.sjf.open.User.Builder();
115.  }
117.  /**
118.   * Creates a new User RecordBuilder by copying an existing Builder.
119.   * @param other The existing builder to copy.
120.   * @return A new User RecordBuilder
121.   */
122.  public static com.sjf.open.User.Builder newBuilder(com.sjf.open.User.Builder other) {
123.    return new com.sjf.open.User.Builder(other);
124.  }
126.  /**
127.   * Creates a new User RecordBuilder by copying an existing User instance.
128.   * @param other The existing instance to copy.
129.   * @return A new User RecordBuilder
130.   */
131.  public static com.sjf.open.User.Builder newBuilder(com.sjf.open.User other) {
132.    return new com.sjf.open.User.Builder(other);
133.  }
135.  /**
136.   * RecordBuilder for User instances.
137.   */
138.  public static class Builder extends org.apache.avro.specific.SpecificRecordBuilderBase<User>
139.    implements org.apache.avro.data.RecordBuilder<User> {
141.    private java.lang.CharSequence name;
142.    private java.lang.Integer age;
143.    private java.lang.CharSequence favorite;
145.    /** Creates a new Builder */
146.    private Builder() {
147.      super(SCHEMA$);
148.    }
150.    /**
151.     * Creates a Builder by copying an existing Builder.
152.     * @param other The existing Builder to copy.
153.     */
154.    private Builder(com.sjf.open.User.Builder other) {
155.      super(other);
156.      if (isValidValue(fields()[0], other.name)) {
157.        this.name = data().deepCopy(fields()[0].schema(), other.name);
158.        fieldSetFlags()[0] = true;
159.      }
160.      if (isValidValue(fields()[1], other.age)) {
161.        this.age = data().deepCopy(fields()[1].schema(), other.age);
162.        fieldSetFlags()[1] = true;
163.      }
164.      if (isValidValue(fields()[2], other.favorite)) {
165.        this.favorite = data().deepCopy(fields()[2].schema(), other.favorite);
166.        fieldSetFlags()[2] = true;
167.      }
168.    }
170.    /**
171.     * Creates a Builder by copying an existing User instance
172.     * @param other The existing instance to copy.
173.     */
174.    private Builder(com.sjf.open.User other) {
175.            super(SCHEMA$);
176.      if (isValidValue(fields()[0], other.name)) {
177.        this.name = data().deepCopy(fields()[0].schema(), other.name);
178.        fieldSetFlags()[0] = true;
179.      }
180.      if (isValidValue(fields()[1], other.age)) {
181.        this.age = data().deepCopy(fields()[1].schema(), other.age);
182.        fieldSetFlags()[1] = true;
183.      }
184.      if (isValidValue(fields()[2], other.favorite)) {
185.        this.favorite = data().deepCopy(fields()[2].schema(), other.favorite);
186.        fieldSetFlags()[2] = true;
187.      }
188.    }
190.    /**
191.      * Gets the value of the 'name' field.
192.      * @return The value.
193.      */
194.    public java.lang.CharSequence getName() {
195.      return name;
196.    }
198.    /**
199.      * Sets the value of the 'name' field.
200.      * @param value The value of 'name'.
201.      * @return This builder.
202.      */
203.    public com.sjf.open.User.Builder setName(java.lang.CharSequence value) {
204.      validate(fields()[0], value);
205.      this.name = value;
206.      fieldSetFlags()[0] = true;
207.      return this;
208.    }
210.    /**
211.      * Checks whether the 'name' field has been set.
212.      * @return True if the 'name' field has been set, false otherwise.
213.      */
214.    public boolean hasName() {
215.      return fieldSetFlags()[0];
216.    }
219.    /**
220.      * Clears the value of the 'name' field.
221.      * @return This builder.
222.      */
223.    public com.sjf.open.User.Builder clearName() {
224.      name = null;
225.      fieldSetFlags()[0] = false;
226.      return this;
227.    }
229.    /**
230.      * Gets the value of the 'age' field.
231.      * @return The value.
232.      */
233.    public java.lang.Integer getAge() {
234.      return age;
235.    }
237.    /**
238.      * Sets the value of the 'age' field.
239.      * @param value The value of 'age'.
240.      * @return This builder.
241.      */
242.    public com.sjf.open.User.Builder setAge(java.lang.Integer value) {
243.      validate(fields()[1], value);
244.      this.age = value;
245.      fieldSetFlags()[1] = true;
246.      return this;
247.    }
249.    /**
250.      * Checks whether the 'age' field has been set.
251.      * @return True if the 'age' field has been set, false otherwise.
252.      */
253.    public boolean hasAge() {
254.      return fieldSetFlags()[1];
255.    }
258.    /**
259.      * Clears the value of the 'age' field.
260.      * @return This builder.
261.      */
262.    public com.sjf.open.User.Builder clearAge() {
263.      age = null;
264.      fieldSetFlags()[1] = false;
265.      return this;
266.    }
268.    /**
269.      * Gets the value of the 'favorite' field.
270.      * @return The value.
271.      */
272.    public java.lang.CharSequence getFavorite() {
273.      return favorite;
274.    }
276.    /**
277.      * Sets the value of the 'favorite' field.
278.      * @param value The value of 'favorite'.
279.      * @return This builder.
280.      */
281.    public com.sjf.open.User.Builder setFavorite(java.lang.CharSequence value) {
282.      validate(fields()[2], value);
283.      this.favorite = value;
284.      fieldSetFlags()[2] = true;
285.      return this;
286.    }
288.    /**
289.      * Checks whether the 'favorite' field has been set.
290.      * @return True if the 'favorite' field has been set, false otherwise.
291.      */
292.    public boolean hasFavorite() {
293.      return fieldSetFlags()[2];
294.    }
297.    /**
298.      * Clears the value of the 'favorite' field.
299.      * @return This builder.
300.      */
301.    public com.sjf.open.User.Builder clearFavorite() {
302.      favorite = null;
303.      fieldSetFlags()[2] = false;
304.      return this;
305.    }
307.    @Override
308.    public User build() {
309.      try {
310.        User record = new User();
311.        record.name = fieldSetFlags()[0] ? this.name : (java.lang.CharSequence) defaultValue(fields()[0]);
312.        record.age = fieldSetFlags()[1] ? this.age : (java.lang.Integer) defaultValue(fields()[1]);
313.        record.favorite = fieldSetFlags()[2] ? this.favorite : (java.lang.CharSequence) defaultValue(fields()[2]);
314.        return record;
315.      } catch (Exception e) {
316.        throw new org.apache.avro.AvroRuntimeException(e);
317.      }
318.    }
319.  }
321.  private static final org.apache.avro.io.DatumWriter
322.    WRITER$ = new org.apache.avro.specific.SpecificDatumWriter(SCHEMA$);
324.  @Override public void writeExternal(java.io.ObjectOutput out)
325.    throws java.io.IOException {
326.    WRITER$.write(this, SpecificData.getEncoder(out));
327.  }
329.  private static final org.apache.avro.io.DatumReader
330.    READER$ = new org.apache.avro.specific.SpecificDatumReader(SCHEMA$);
332.  @Override public void readExternal(java.io.ObjectInput in)
333.    throws java.io.IOException {
334.    READER$.read(this, SpecificData.getDecoder(in));
335.  }
337. }

4. 创建User对象

既然我们已经完成了代码自动生成，那我们创建一些用户，序列化到磁盘上的文件中，然后再读取文件，反序列化为用户对象。

1. // 方式一
2. User user1 = new User();
3. user1.setName("yoona");
4. user1.setAge(25);
5. user1.setFavorite("football");
7. // 方式二
8. User user2 = new User("sjf", 24, "basketball");
10. // 方式三
11. User user3 = User.newBuilder().setName("Tom").setAge(26).setFavorite("badminton").build();

如这个例子所示，Avro对象可以通过调用构造函数直接创建用户或通过使用builder来创建用户。不同于构造函数，Builder的方式会把Schema设置的默认值赋给对象相应的属性（Unlike constructors, builders will automatically set any default values specified in the schema）。除此之外，Builder会校验数据判断是否符合schema，但是构造器不会做这些校验工作，直到对象被序列化的时候才会校验。但是通常使用构造函数会有更好的性能，因为Builder在写数据之前会创建数据的副本（However, using constructors directly generally offers better performance, as builders create a copy of the datastructure before it is written）。

5. 序列化操作

将上述几个User进行序列化操作，并将序列化的数据存放到user.avro文件中：

运行上述代码之后，将会在磁盘产生user.avro文件，里面是avro序列化user的数据。

我们创建一个DatumWriter，将Java对象转换成内存中的序列化格式。

6. 反序列化操作

我们可以对上述代码序列化之后的数据user.avro进行反序列化为user对象：

1.   /**
2.     * 反序列化
3.     */
4.    public static void deserialize(){
5.        DatumReader<User> reader = new SpecificDatumReader<User>(User.class);
6.        DataFileReader<User> fileReader = null;
7.        try {
8.            fileReader = new DataFileReader<User>(new File(
9.                    "user.avro"), reader);
10.            User user = null;
11.            while (fileReader.hasNext()) {
12.                // 复用user对象，避免重复分配内存和GC
13.                user = fileReader.next(user);
14.                System.out.println(user);
15.            }
16.        } catch (IOException e) {
17.            e.printStackTrace();
18.        }
19.    }

输出内容：

1. {"name": "yoona", "age": 25, "favorite": "football"}
2. {"name": "sjf", "age": 24, "favorite": "basketball"}
3. {"name": "Tom", "age": 26, "favorite": "football"}

7. 完整例子

1. import org.apache.avro.file.DataFileReader;
2. import org.apache.avro.file.DataFileWriter;
3. import org.apache.avro.io.DatumReader;
4. import org.apache.avro.io.DatumWriter;
5. import org.apache.avro.specific.SpecificDatumReader;
6. import org.apache.avro.specific.SpecificDatumWriter;
8. import java.io.File;
9. import java.io.IOException;
11. /**
12. * Created by xiaosi on 16-9-14.
13. */
14. public class Test {
16.    /**
17.     * 序列化
18.     */
19.    public static void serialize(){
20.        // 方式一
21.        User user1 = new User();
22.        user1.setName("yoona");
23.        user1.setAge(25);
24.        user1.setFavorite("football");
26.        // 方式二
27.        User user2 = new User("sjf", 24, "basketball");
29.        // 方式三
30.        User user3 = User.newBuilder().setName("Tom").setAge(26).build();
32.        // 序列化
33.        DatumWriter<User> userDatumWriter = new SpecificDatumWriter<User>(User.class);
34.        DataFileWriter<User> dataFileWriter = new DataFileWriter<User>(userDatumWriter);
35.        try {
36.            dataFileWriter.create(user1.getSchema(), new File("user.avro"));
37.            dataFileWriter.append(user1);
38.            dataFileWriter.append(user2);
39.            dataFileWriter.append(user3);
40.            dataFileWriter.close();
41.        } catch (IOException e) {
42.            e.printStackTrace();
43.        }
44.    }
46.    /**
47.     * 反序列化
48.     */
49.    public static void deserialize(){
50.        DatumReader<User> reader = new SpecificDatumReader<User>(User.class);
51.        DataFileReader<User> fileReader = null;
52.        try {
53.            fileReader = new DataFileReader<User>(new File(
54.                    "user.avro"), reader);
55.            User user = null;
56.            while (fileReader.hasNext()) {
57.                // 复用user对象，避免重复分配内存和GC
58.                user = fileReader.next(user);
59.                System.out.println(user);
60.            }
61.        } catch (IOException e) {
62.            e.printStackTrace();
63.        }
64.    }
66.    public static void main(String[] args) {
67.        serialize();
68.        deserialize();
69.    }
70. }