Apache Avro for Java

概要

Apache Avroは、Apache Software Foundationが開発したデータシリアライゼーションシステムです。JSON形式で定義されたスキーマを使用し、コンパクトなバイナリフォーマットでデータをシリアライズします。特にスキーマ進化のサポートが優れており、書き込み時と読み込み時で異なるスキーマを使用できるため、データの長期保存や大規模システムでのデータ交換に適しています。

主な特徴

1. 豊富なデータ型サポート: プリミティブ型（null, boolean, int, long, float, double, bytes, string）と複合型（record, enum, array, map, union, fixed）
2. 論理型: decimal, date, time-millis, timestamp-millis, uuidなどの意味的な型
3. 効率的なエンコーディング: 可変長エンコーディングを使用し、コンパクトなバイナリ形式
4. スキーマ互換性: 読み書きスキーマ間の自動変換
5. コード生成: 型安全なJavaクラスの自動生成

インストール

Maven

<dependency>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro</artifactId>
    <version>1.11.3</version>
</dependency>

<!-- コード生成用プラグイン -->
<plugin>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro-maven-plugin</artifactId>
    <version>1.11.3</version>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <goals>
                <goal>schema</goal>
            </goals>
            <configuration>
                <sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory>
                <outputDirectory>${project.basedir}/src/main/java/</outputDirectory>
            </configuration>
        </execution>
    </executions>
</plugin>

Gradle

dependencies {
    implementation 'org.apache.avro:avro:1.11.3'
}

// コード生成用プラグイン
plugins {
    id "com.github.davidmc24.gradle.plugin.avro" version "1.8.0"
}

基本的な使い方

スキーマ定義

{
  "namespace": "com.example.avro",
  "type": "record",
  "name": "User",
  "fields": [
    {
      "name": "id",
      "type": "long"
    },
    {
      "name": "name",
      "type": "string"
    },
    {
      "name": "email",
      "type": ["null", "string"],
      "default": null
    },
    {
      "name": "age",
      "type": "int"
    },
    {
      "name": "roles",
      "type": {
        "type": "array",
        "items": "string"
      }
    },
    {
      "name": "metadata",
      "type": {
        "type": "map",
        "values": "string"
      }
    },
    {
      "name": "createdAt",
      "type": {
        "type": "long",
        "logicalType": "timestamp-millis"
      }
    }
  ]
}

Generic APIの使用

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.*;

public class GenericExample {
    public static void main(String[] args) throws Exception {
        // スキーマの読み込み
        Schema schema = new Schema.Parser().parse(
            new File("src/main/avro/user.avsc")
        );
        
        // レコードの作成
        GenericRecord user = new GenericData.Record(schema);
        user.put("id", 1L);
        user.put("name", "田中太郎");
        user.put("email", "[email protected]");
        user.put("age", 30);
        user.put("roles", Arrays.asList("admin", "user"));
        
        Map<String, String> metadata = new HashMap<>();
        metadata.put("department", "engineering");
        user.put("metadata", metadata);
        user.put("createdAt", System.currentTimeMillis());
        
        // シリアライズ
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        DatumWriter<GenericRecord> writer = new GenericDatumWriter<>(schema);
        Encoder encoder = EncoderFactory.get().binaryEncoder(out, null);
        writer.write(user, encoder);
        encoder.flush();
        
        byte[] serialized = out.toByteArray();
        System.out.println("シリアライズサイズ: " + serialized.length);
        
        // デシリアライズ
        DatumReader<GenericRecord> reader = new GenericDatumReader<>(schema);
        Decoder decoder = DecoderFactory.get().binaryDecoder(serialized, null);
        GenericRecord result = reader.read(null, decoder);
        
        System.out.println("デコード結果: " + result);
    }
}

Specific APIの使用（コード生成後）

import com.example.avro.User;
import org.apache.avro.io.*;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.avro.specific.SpecificDatumWriter;

public class SpecificExample {
    public static void main(String[] args) throws Exception {
        // Builderパターンでインスタンス作成
        User user = User.newBuilder()
            .setId(1L)
            .setName("田中太郎")
            .setEmail("[email protected]")
            .setAge(30)
            .setRoles(Arrays.asList("admin", "user"))
            .setMetadata(Map.of(
                "department", "engineering",
                "level", "senior"
            ))
            .setCreatedAt(System.currentTimeMillis())
            .build();
        
        // シリアライズ
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        DatumWriter<User> writer = new SpecificDatumWriter<>(User.class);
        Encoder encoder = EncoderFactory.get().binaryEncoder(out, null);
        writer.write(user, encoder);
        encoder.flush();
        
        byte[] serialized = out.toByteArray();
        
        // デシリアライズ
        DatumReader<User> reader = new SpecificDatumReader<>(User.class);
        Decoder decoder = DecoderFactory.get().binaryDecoder(serialized, null);
        User result = reader.read(null, decoder);
        
        System.out.println("名前: " + result.getName());
        System.out.println("メール: " + result.getEmail());
    }
}

Reflect APIの使用

import org.apache.avro.Schema;
import org.apache.avro.reflect.ReflectData;
import org.apache.avro.reflect.ReflectDatumReader;
import org.apache.avro.reflect.ReflectDatumWriter;

// 既存のPOJOクラス
public class Employee {
    private long id;
    private String name;
    private String department;
    private double salary;
    
    // getter/setter省略
}

public class ReflectExample {
    public static void main(String[] args) throws Exception {
        Employee emp = new Employee();
        emp.setId(1L);
        emp.setName("佐藤花子");
        emp.setDepartment("開発部");
        emp.setSalary(500000);
        
        // スキーマの自動生成
        Schema schema = ReflectData.get().getSchema(Employee.class);
        
        // シリアライズ
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        DatumWriter<Employee> writer = new ReflectDatumWriter<>(Employee.class);
        Encoder encoder = EncoderFactory.get().binaryEncoder(out, null);
        writer.write(emp, encoder);
        encoder.flush();
        
        // デシリアライズ
        DatumReader<Employee> reader = new ReflectDatumReader<>(Employee.class);
        Decoder decoder = DecoderFactory.get().binaryDecoder(out.toByteArray(), null);
        Employee result = reader.read(null, decoder);
        
        System.out.println("復元結果: " + result.getName());
    }
}

スキーマ進化

後方互換性のある変更

// 旧スキーマ
{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "id", "type": "long"},
    {"name": "name", "type": "string"}
  ]
}

// 新スキーマ（後方互換）
{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "id", "type": "long"},
    {"name": "name", "type": "string"},
    {"name": "email", "type": ["null", "string"], "default": null},
    {"name": "age", "type": "int", "default": 0}
  ]
}

スキーマ解決の例

public class SchemaEvolutionExample {
    public static void main(String[] args) throws Exception {
        // 書き込みスキーマ（旧）
        Schema writerSchema = new Schema.Parser().parse(
            "{\"type\":\"record\",\"name\":\"User\",\"fields\":[" +
            "{\"name\":\"id\",\"type\":\"long\"}," +
            "{\"name\":\"name\",\"type\":\"string\"}]}"
        );
        
        // 読み込みスキーマ（新）
        Schema readerSchema = new Schema.Parser().parse(
            "{\"type\":\"record\",\"name\":\"User\",\"fields\":[" +
            "{\"name\":\"id\",\"type\":\"long\"}," +
            "{\"name\":\"name\",\"type\":\"string\"}," +
            "{\"name\":\"email\",\"type\":[\"null\",\"string\"],\"default\":null}," +
            "{\"name\":\"age\",\"type\":\"int\",\"default\":0}]}"
        );
        
        // 旧スキーマでデータ作成
        GenericRecord oldUser = new GenericData.Record(writerSchema);
        oldUser.put("id", 1L);
        oldUser.put("name", "鈴木一郎");
        
        // シリアライズ
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        DatumWriter<GenericRecord> writer = new GenericDatumWriter<>(writerSchema);
        Encoder encoder = EncoderFactory.get().binaryEncoder(out, null);
        writer.write(oldUser, encoder);
        encoder.flush();
        
        // 新スキーマでデシリアライズ
        DatumReader<GenericRecord> reader = new GenericDatumReader<>(
            writerSchema, readerSchema
        );
        Decoder decoder = DecoderFactory.get().binaryDecoder(out.toByteArray(), null);
        GenericRecord newUser = reader.read(null, decoder);
        
        System.out.println("ID: " + newUser.get("id"));
        System.out.println("名前: " + newUser.get("name"));
        System.out.println("メール: " + newUser.get("email")); // null (デフォルト値)
        System.out.println("年齢: " + newUser.get("age")); // 0 (デフォルト値)
    }
}

高度な機能

ファイルへのシリアライズ

import org.apache.avro.file.DataFileReader;
import org.apache.avro.file.DataFileWriter;

public class FileExample {
    public static void writeToFile(List<User> users, String filename) 
            throws IOException {
        DatumWriter<User> userDatumWriter = new SpecificDatumWriter<>(User.class);
        DataFileWriter<User> dataFileWriter = new DataFileWriter<>(userDatumWriter);
        
        // スキーマとコーデックの設定
        dataFileWriter.create(User.getClassSchema(), new File(filename));
        
        // データの書き込み
        for (User user : users) {
            dataFileWriter.append(user);
        }
        
        dataFileWriter.close();
    }
    
    public static List<User> readFromFile(String filename) 
            throws IOException {
        List<User> users = new ArrayList<>();
        
        DatumReader<User> userDatumReader = new SpecificDatumReader<>(User.class);
        DataFileReader<User> dataFileReader = new DataFileReader<>(
            new File(filename), userDatumReader
        );
        
        while (dataFileReader.hasNext()) {
            users.add(dataFileReader.next());
        }
        
        dataFileReader.close();
        return users;
    }
}

コード生成のカスタマイズ

import org.apache.avro.compiler.specific.SpecificCompiler;

public class CodeGenerationCustomization {
    public static void customizeGeneration() {
        SpecificCompiler compiler = new SpecificCompiler(schema);
        
        // String型の設定（デフォルトはUtf8）
        compiler.setStringType(StringType.String);
        
        // フィールドの可視性
        compiler.setFieldVisibility(FieldVisibility.PRIVATE);
        
        // Setterメソッドの生成
        compiler.setCreateSetters(true);
        
        // Optional getterの使用
        compiler.setGettersReturnOptional(true);
        
        // コード生成
        compiler.compileToDestination(null, new File("src/main/java"));
    }
}

パフォーマンス最適化

public class PerformanceOptimization {
    // バッファリングを使用した高速エンコーディング
    public static byte[] fastEncode(User user) throws IOException {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        BinaryEncoder encoder = EncoderFactory.get()
            .blockingBinaryEncoder(out, null);
        
        DatumWriter<User> writer = new SpecificDatumWriter<>(User.class);
        writer.write(user, encoder);
        encoder.flush();
        
        return out.toByteArray();
    }
    
    // ダイレクトバイナリエンコーディング
    public static byte[] directEncode(User user) throws IOException {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        Encoder encoder = EncoderFactory.get()
            .directBinaryEncoder(out, null);
        
        DatumWriter<User> writer = new SpecificDatumWriter<>(User.class);
        writer.write(user, encoder);
        
        return out.toByteArray();
    }
}

ベストプラクティス

1. APIの選択

• Specific API: 型安全性とパフォーマンスを重視する場合
• Generic API: スキーマが動的に変わる場合
• Reflect API: 既存のPOJOを使用したい場合

2. スキーマ設計

// 良い例: デフォルト値を持つオプショナルフィールド
{
  "name": "email",
  "type": ["null", "string"],
  "default": null
}

// 避けるべき: デフォルト値のない新フィールド
{
  "name": "newField",
  "type": "string"
  // デフォルト値がないと後方互換性が失われる
}

3. パフォーマンスの最適化

• 文字列操作が多い場合はStringType.Stringを使用
• 大量のデータを扱う場合はBufferedBinaryEncoderを使用
• スキーマのキャッシュを使用してパースのオーバーヘッドを削減

他のシリアライゼーション形式との比較

Avro vs Protocol Buffers

特性	Avro	Protocol Buffers
スキーマ定義	JSON	.protoファイル
スキーマ進化	優れている	限定的
動的スキーマ	サポート	非サポート
データサイズ	コンパクト	よりコンパクト
Hadoop統合	ネイティブ	追加実装必要

Avro vs JSON

特性	Avro	JSON
サイズ	コンパクト	大きい
速度	高速	標準的
スキーマ	必須	オプショナル
可読性	バイナリ	テキスト
型情報	豊富	限定的

まとめ

Apache Avroは、特にデータの長期保存や大規模システムでのデータ交換に優れたシリアライゼーションフォーマットです。スキーマ進化のサポートにより、システムの成長に対応しやすく、Hadoopエコシステムとの統合も良好です。JavaではSpecific APIを使用することで、型安全性とパフォーマンスの両方を得ることができます。