在本文中，我们将介绍如何使用Java API在Hadoop中读写Avro文件。

所需的Jar包

要编写Java程序来读写Avro文件，我们需要将以下jar文件放在classpath中。我们可以将它们添加为Maven依赖项或者复制jar。

• avro-1.8.2.jar
• avro-tools-1.8.2.jar
• Hymanson-mapper-asl-1.9.13.jar
• Hymanson-core-asl-1.9.13.jar
• slf4j-api-1.7.25.jar

Java程序编写avro文件

由于使用了Avro，因此我们需要avro模式。

schema.avsc

{
  "type":	"record",
  "name":	"EmployeeRecord",
  "doc":	"employee records",
  "fields": 
    [{
      "name":	"id",	
      "type":	"int"
    }, 
    {
      "name":	"empName",
      "type":	"string"
    }, 
    {
      "name":	"age",
      "type":	"int"
    }
  ]
}

Java代码

import java.io.File;
import java.io.IOException;
import org.apache.avro.Schema;
import org.apache.avro.file.DataFileWriter;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.DatumWriter;

public class ExampleAvroWriter {

  public static void main(String[] args) {
    Schema schema = parseSchema();
    writeToAvroFile(schema);
  }
	
  // Method to parse the schema
  private static Schema parseSchema() {
    Schema.Parser parser = new Schema.Parser();
    Schema schema = null;
    try {
      // Path to schema file
      schema = parser.parse(ClassLoader.getSystemResourceAsStream("resources/schema.avsc"));			
    } catch (IOException e) {
      e.printStackTrace();			
    }
    return schema;		
  }
		
  private static void writeToAvroFile(Schema schema) {
    GenericRecord emp1 = new GenericData.Record(schema);
    emp1.put("id", 1);
    emp1.put("empName", "Batista");
    emp1.put("age", 45);
    
    GenericRecord emp2 = new GenericData.Record(schema);
    emp2.put("id", 2);
    emp2.put("empName", "Jigmi");
    emp2.put("age", 23);
          
    DatumWriter datumWriter = new GenericDatumWriter(schema);
    DataFileWriter dataFileWriter = null;
    try {
      
      // Local File system - out file path
      File file = new File("/home/theitroad/emp.avro");
      
      dataFileWriter = new DataFileWriter(datumWriter);
      // for compression
      //dataFileWriter.setCodec(CodecFactory.snappyCodec());
      dataFileWriter.create(schema, file);
      
      dataFileWriter.append(emp1);
      dataFileWriter.append(emp2);
    } catch (IOException e) {
      // TODO Auto-generated catch block
      e.printStackTrace();
    }finally {
      if(dataFileWriter != null) {
        try {
          dataFileWriter.close();
        } catch (IOException e) {
          // TODO Auto-generated catch block
          e.printStackTrace();
        }
      }			
    }	        
  }
}

请注意，在此代码输出中，avro文件是在本地文件系统中创建的。如果要在HDFS中创建输出文件，则需要使用以下更改来传递路径。

// For HDFS - out file path
Configuration conf = new Configuration();
FileSystem fs = FileSystem.get(URI.create("hdfs://hostname:port/user/out/emp.avro"), conf);
OutputStream out = fs.create(newPath("hdfs://hostname:port/user/out/emp.avro"));

并在create方法中传递此OutputStream对象

dataFileWriter.create(schema, out);

在Hadoop环境中执行程序

在Hadoop环境中运行该程序之前，我们需要将上述jars放入$ HADOOP_INSTALLATION_DIR / share / hadoop / mapreduce / lib中。

如果版本不匹配，还将当前版本的Avro-1.x.x jar放在$ HADOOP_INSTALLATION_DIR / share / hadoop / common / lib位置。
要在Hadoop环境中执行上述Java程序，我们需要在Hadoop的类路径中添加包含Java程序的.class文件的目录。

export HADOOP_CLASSPATH='/huser/eclipse-workspace/theitroad/bin'

我的exampleAvroWriter.class文件位于/ huser / eclipse-workspace / theitroad / bin位置，因此我已经导出了该路径。

然后，我们可以使用以下命令运行该程序-

$ hadoop org.theitroad.ExampleAvroWriter

Java程序读取avro文件

为了读取上一个示例中存储在HDFS中的avro文件，可以使用以下方法。根据配置提供HOSTNAME和PORT的值。

private static void readFromAvroFile(Schema schema) {		
  Configuration conf = new	Configuration();
  DataFileReader dataFileReader = null;
  try {
    FsInput in = new FsInput(new Path("hdfs://HOSTNAME:PORT/user/out/emp.avro"), conf);
    DatumReader datumReader = new GenericDatumReader(schema);
    dataFileReader = new DataFileReader(in, datumReader);
    GenericRecord emp = null;
    while (dataFileReader.hasNext()) {
      emp = dataFileReader.next(emp);
      System.out.println(emp);
    }
  }catch (IOException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
  }finally {
    if(dataFileReader != null) {
      try {
        dataFileReader.close();
      } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
      }
    }			
  }	   
}

输出量

{"id": 1, "empName": "Batista", "age": 45}
{"id": 2, "empName": "Jigmi", "age": 23}

如果要从本地文件系统读取avro文件，则可以使用以下方法。

private static void readFromAvroFile(Schema schema) {
  DataFileReader dataFileReader = null;
  try {
    File file = new File("/home/theitroad/emp.avro");
    DatumReader datumReader = new GenericDatumReader(schema);
    dataFileReader = new DataFileReader(file, datumReader);
    GenericRecord emp = null;
    while (dataFileReader.hasNext()) {
      emp = dataFileReader.next(emp);
      System.out.println(emp);
    }
  }catch (IOException e) {
    // TODO Auto-generated catch block
    e.printStackTrace();
  }finally {
    if(dataFileReader != null) {
      try {
        dataFileReader.close();
      } catch (IOException e) {
        // TODO Auto-generated catch block
        e.printStackTrace();
      }
    }
  }	   
}