[하둡에코시스템] Apache Avro / 아파치 에이브로

아파치 에이브로란 ? 

https://dennyglee.com/2013/03/12/using-avro-with-hdinsight-on-azure-at-343-industries/

- 특정 언어에 종속되지 않는 언어 중립적 데이터 직렬화 시스템

- 하둡 Writable의 주요 단점인 언어 이식성 해결 위해 생겨남

 

아파치 쓰리프트, 구글 프로토콜 버퍼와 다른 차별화된 특성가지고 있음

 

데이터는 다른 시스템과 비슷하게 언어 독립 스키마로 기술됨

에이브로에서 코드 생성은 선택사항임

데이터를 읽고 쓰는 시점에 스키마는 항상 존재한다 가정함 - 매우 간결한 코딩이 가능

 

스키마의 작성

JSON

데이터는 바이너리 포맷으로 인코딩

 

에이브로 명세 - 모든 구현체가 지원해야 하는 버이너리 포맷에 대한 자세한 내용

API - 에이브로 명세에서 빠져있는 내용임. 각 특정언어에 따라 다르게 작성됨. 언어의 바인딩 편의성 높이고 상호운영성 저하 문제 해결됨

 

스키마해석 - 신중하게 정의된 어떠한 제약조건에서도 데이터를 읽는 데 사용되는 스키마와 데이터를 쓰는 데 사용되는 스키마가 같지 않아도 된다.(스키마 변형 메커니즘)

 

ex ) 과거의 데이터를 읽을 때 사용한 스키마에 새로운 필드를 추가할 수 있다. 새로운 사용자와 기존 사용자는 모두 과거의 데이터를 문제없이 읽을 수 있으며새로운 사용자는 새로운 필드가 추가된 데이터를 쓸수 있다. 기존 사용자는 새로운 데이터를 보게되는데 새로운 필드는 무시하고 기존 데이터 작업처럼 처리할 수 있다.

 

객체 컨테이너 포맷 제공(하둡 시퀀스 파일과 유사함)

에이브로 데이터 파일은 스키마가 저장된 메타데이터 섹션을 포함하고 있어 자신을 설명하는 파일임

에이브로 데이터 파일은 압축과 분할 기능 제공

 

에이브로 자료형과 스키마

에이브로의 기본 자료형 표시

{"type":"null"}
{"type":"boolean"}
{"type":"int"}
{"type":"long"}
{"type":"float"}
{"type":"double"}
{"type":"bytes"}
{"type":"string"}

에이브로 복합 자료형

array 순서있는 객체 집합, 동일 타입

{
 "type": "array",
 "items": "long"
}
 

map 순서 없는 키-값, 동일 타입

{
 "type": "map",
 "values": "string"
}

record 임의의 자료형

{
 "type": "record",
 "name": "WeatherRecord",
 "doc": "A weather reading.",
 "fields": [
 {"name": "year", "type": "int"},
 {"name": "temperature", "type": "int"},
 {"name": "stationId", "type": "string"}
 ]
}
 

enum 명명된 값의 집합

{
 "type": "enum",
 "name": "Cutlery",
 "doc": "An eating utensil.",
 "symbols": ["KNIFE", "FORK", "SPOON"]
}
 

fixed 고정길이 8비트 부호 없는 바이트

{
 "type": "fixed",
 "name": "Md5Hash",
 "size": 16
}
 

union 스키마의 유니온, 배열의 각요소는 스키마임

[
 "null",
 "string",
 {"type": "map", "values": "string"}
]

 

에이브로 자료형과 다른 프로그래밍 언어 자료형 매핑 필요

 

자바 - 제네릭 매핑 : 스키마를 결정할 수 없을 때

자바, C++ - 구체적 매핑 : 스키마 데이터 표현 코드 생성

자바 - 리플렉트 매핑 : 에이브로 자료형을 기존 자바 자료형으로 매핑 

 

 

인메모리 직렬화와 역직렬화

에이브로 스키마 예시 - 해당 에이브로 스키마는 StringPair.avsc 에 저장됨 

{
 "type": "record",
 "name": "StringPair",
 "doc": "A pair of strings.",
 "fields": [
 {"name": "left", "type": "string"},
 {"name": "right", "type": "string"}
 ]
}

파일을 클래스 경로에 저장한 후 로딩함

Schema.Parser parser = new Schema.Parser();
Schema schema = parser.parse(
getClass().getResourceAsStream("StringPair.avsc"));

제네릭 API 를 사용하여 에이브로 레코드의 인스턴스를 생성함

GenericRecord datum = new GenericData.Record(schema);
datum.put("left", "L");
datum.put("right", "R");

출력 스트림에 레코드를 직렬화함

ByteArrayOutputStream out = new ByteArrayOutputStream();
DatumWriter<GenericRecord> writer =
new GenericDatumWriter<GenericRecord>(schema);
Encoder encoder = EncoderFactory.get().binaryEncoder(out, null);
writer.write(datum, encoder);
encoder.flush();
out.close();

DatumReader : 데이터 객체를 인코더가 이해할 수 있는 자료형으로 반환

GenericDatumReader : GenericRecord의 필드를 인코더로 전달

이전의 생성된 인코더를 재사용하지 않기 때문에 인코더 팩토리에 null 전달

writer() 스트림 닫기 전에 필요하면 더 호출 가능

encoder.flush(); write메서드 호출 후 인코더 플러시하고 출력 스트림 닫음

 

해당 과정 반대로 하면 바이트 버퍼에서 객체 읽을 수 있음

 DatumReader<GenericRecord> reader =
 new GenericDatumReader<GenericRecord>(schema);
 Decoder decoder = DecoderFactory.get().binaryDecoder(out.toByteArray(),
 null);
 GenericRecord result = reader.read(null, decoder);
 assertThat(result.get("left").toString(), is("L"));
 assertThat(result.get("right").toString(), is("R"));

 

 

구체적인 API

메이븐에 추가하여 자바로된 스키마 코드 생성할 수 있다.

<project>
 ...
 <build>
 <plugins>
 <plugin>
 <groupId>org.apache.avro</groupId>
 <artifactId>avro-maven-plugin</artifactId>
 <version>${avro.version}</version>
 <executions>
 <execution>
 <id>schemas</id>
 <phase>generate-sources</phase>
 <goals>
 <goal>schema</goal>
 </goals>
 <configuration>
 <includes>
 <include>StringPair.avsc</include>
 </includes>
 <stringType>String</stringType>
 <sourceDirectory>src/main/resources</sourceDirectory>
 <outputDirectory>${project.build.directory}/generated-sources/java
 </outputDirectory>
 </configuration>
 </execution>
 </executions>
 </plugin>
 </plugins>
 </build>
 ...
</project>
 

에이브로 데이터 파일

인메모리 스트림에서 파일 읽기

데이터 파일 = 에이브로 스키마 +  헤더 (메타데이터 (싱크마커 포함)) +일련의 블록 (직렬화된 에이브로 객체가 있는)

데이터 파일에 기록된 객체는 반드시 파일의 스키마와 일치해야한다.

일치하지않는 경우에 append시 예외 발생할 수 있다.

File file = new File("data.avro");
DatumWriter<GenericRecord> writer =
 new GenericDatumWriter<GenericRecord>(schema);
DataFileWriter<GenericRecord> dataFileWriter =
new DataFileWriter<GenericRecord>(writer);
dataFileWriter.create(schema, file);
dataFileWriter.append(datum);
dataFileWriter.close();
 

 

파일에 포함된 메타데이터 참조하여 읽기 때문에 스키마를 따로 정의하지않아도됨

getSchema()를 이용하면 DataFileReader 인스턴스의 스키마 정보 얻을 수 있고 원본 객체에 사용한 스키마와 같은지 확인가능 

DatumReader<GenericRecord> reader = new GenericDatumReader<GenericRecord>();
DataFileReader<GenericRecord> dataFileReader =
new DataFileReader<GenericRecord>(file, reader);
assertThat("Schema is the same", schema, is(dataFileReader.getSchema()));

DataFileReader는 정규 자바 반복자로 hasNext, next메서드를 반복적으로 호출하여

모든 데이터 객체를 순환할 수 있다. 레코드가 한개만 있는지 확인하고 기대한 필드값 있는 지 확인

assertThat(dataFileReader.hasNext(), is(true));
GenericRecord result = dataFileReader.next();
assertThat(result.get("left").toString(), is("L"));
assertThat(result.get("right").toString(), is("R"));
assertThat(dataFileReader.hasNext(), is(false));

상호 운영성

파이썬 API

import os
import string
import sys
from avro import schema
from avro import io
from avro import datafile

 

에이브로 도구

% java -jar $AVRO_HOME/avro-tools-*.jar tojson pairs.avro

 

스키마 해석

기록할 때 사용한 writer 스키마와 다른 reader의 스키마를 사용하여 데이터를 다시 읽을 수 있다,.

추가된 필드 - reader 신규일 때, reader는 신규 필드의 기본값이용

추가된 필드 - writer 신규일 때, reader는 신규 필드 모르기 때문에 무시함

제거된 필드 - reader 신규일 때, 삭제된 필드 무시

제거된 필드 - writer 신규일 때, 제거된 필드 기록하지 않음. reader 의 스키마를 writer 스키마와 같게 맞추거나 이전으로 갱신함

정렬 순서

record를 제외안 모든 자료형에는 순서가 정해져있음

record는 order 속성 명시하여 정렬 제어할 수 있다.

오름차순 (기본)

내림차순 : descending

무시 : ignore

 

에이브로 맵리듀스

날씨 레코드 

{
 "type": "record",
 "name": "WeatherRecord",
 "doc": "A weather reading.",
 "fields": [
 {"name": "year", "type": "int"},
 {"name": "temperature", "type": "int"},
 {"name": "stationId", "type": "string"}
 ]
}

최고 기온을 찾는 맵리듀스 프로그램, 에이브로 출력 만듦

public class AvroGenericMaxTemperature extends Configured implements Tool {
 
 private static final Schema SCHEMA = new Schema.Parser().parse(
 "{" +
 " \"type\": \"record\"," +
 " \"name\": \"WeatherRecord\"," +
 " \"doc\": \"A weather reading.\"," +
 " \"fields\": [" +
 " {\"name\": \"year\", \"type\": \"int\"}," +
 " {\"name\": \"temperature\", \"type\": \"int\"}," +
 " {\"name\": \"stationId\", \"type\": \"string\"}" +
 " ]" +
 "}"
 );
 public static class MaxTemperatureMapper
 extends Mapper<LongWritable, Text, AvroKey<Integer>,
 AvroValue<GenericRecord>> {
 private NcdcRecordParser parser = new NcdcRecordParser();
 private GenericRecord record = new GenericData.Record(SCHEMA);
 @Override
 protected void map(LongWritable key, Text value, Context context)
 throws IOException, InterruptedException {
 parser.parse(value.toString());
 if (parser.isValidTemperature()) {
 record.put("year", parser.getYearInt());
 record.put("temperature", parser.getAirTemperature());
 record.put("stationId", parser.getStationId());
 context.write(new AvroKey<Integer>(parser.getYearInt()),
 new AvroValue<GenericRecord>(record));
 }
 }
 }
 
 public static class MaxTemperatureReducer
 extends Reducer<AvroKey<Integer>, AvroValue<GenericRecord>,
 AvroKey<GenericRecord>, NullWritable> {
 @Override
 protected void reduce(AvroKey<Integer> key, Iterable<AvroValue<GenericRecord>>
 values, Context context) throws IOException, InterruptedException {
 GenericRecord max = null;
 for (AvroValue<GenericRecord> value : values) {
 GenericRecord record = value.datum();
 if (max == null ||
360 | Chapter 12: Avro (Integer) record.get("temperature") > (Integer) max.get("temperature")) {
 max = newWeatherRecord(record);
 }
 }
 context.write(new AvroKey(max), NullWritable.get());
 }
 private GenericRecord newWeatherRecord(GenericRecord value) {
 GenericRecord record = new GenericData.Record(SCHEMA);
 record.put("year", value.get("year"));
 record.put("temperature", value.get("temperature"));
 record.put("stationId", value.get("stationId"));
 return record;
 }
 }
 @Override
 public int run(String[] args) throws Exception {
 if (args.length != 2) {
 System.err.printf("Usage: %s [generic options] <input> <output>\n",
 getClass().getSimpleName());
 ToolRunner.printGenericCommandUsage(System.err);
 return -1;
 }
 Job job = new Job(getConf(), "Max temperature");
 job.setJarByClass(getClass());
 job.getConfiguration().setBoolean(
 Job.MAPREDUCE_JOB_USER_CLASSPATH_FIRST, true);
 FileInputFormat.addInputPath(job, new Path(args[0]));
 FileOutputFormat.setOutputPath(job, new Path(args[1]));
 AvroJob.setMapOutputKeySchema(job, Schema.create(Schema.Type.INT));
 AvroJob.setMapOutputValueSchema(job, SCHEMA);
 AvroJob.setOutputKeySchema(job, SCHEMA);
 job.setInputFormatClass(TextInputFormat.class);
 job.setOutputFormatClass(AvroKeyOutputFormat.class);
 job.setMapperClass(MaxTemperatureMapper.class);
 job.setReducerClass(MaxTemperatureReducer.class);
 return job.waitForCompletion(true) ? 0 : 1;
 }
 
 public static void main(String[] args) throws Exception {
 int exitCode = ToolRunner.run(new AvroGenericMaxTemperature(), args);
 System.exit(exitCode);
 }
}

프로그램 실행

% export HADOOP_CLASSPATH=avro-examples.jar
% export HADOOP_USER_CLASSPATH_FIRST=true # override version of Avro in Hadoop
% hadoop jar avro-examples.jar AvroGenericMaxTemperature \
 input/ncdc/sample.txt output

결과물출력

% java -jar $AVRO_HOME/avro-tools-*.jar tojson output/part-r-00000.avro
{"year":1949,"temperature":111,"stationId":"012650-99999"}
{"year":1950,"temperature":22,"stationId":"011990-99999"}

 

에이브로 맵리듀스 이용해 정렬

에이브로 데이터 파일을 정렬하는 맵리듀스 프로그램

public class AvroSort extends Configured implements Tool {
 static class SortMapper<K> extends Mapper<AvroKey<K>, NullWritable,
 AvroKey<K>, AvroValue<K>> {
 @Override
 protected void map(AvroKey<K> key, NullWritable value,
 Context context) throws IOException, InterruptedException {
 context.write(key, new AvroValue<K>(key.datum()));
 }
 }
 static class SortReducer<K> extends Reducer<AvroKey<K>, AvroValue<K>,
 AvroKey<K>, NullWritable> {
 @Override
 protected void reduce(AvroKey<K> key, Iterable<AvroValue<K>> values,
 Context context) throws IOException, InterruptedException {
 for (AvroValue<K> value : values) {
 context.write(new AvroKey(value.datum()), NullWritable.get());
 }
 }
 }
 @Override
 public int run(String[] args) throws Exception {
 
 if (args.length != 3) {
 System.err.printf(
 "Usage: %s [generic options] <input> <output> <schema-file>\n",
 getClass().getSimpleName());
 ToolRunner.printGenericCommandUsage(System.err);
 return -1;
 }
 
 String input = args[0];
 String output = args[1];
 String schemaFile = args[2];
 Job job = new Job(getConf(), "Avro sort");
 job.setJarByClass(getClass());
 job.getConfiguration().setBoolean(Job.MAPREDUCE_JOB_USER_CLASSPATH_FIRST, true);
 FileInputFormat.addInputPath(job, new Path(input));
 FileOutputFormat.setOutputPath(job, new Path(output));
 AvroJob.setDataModelClass(job, GenericData.class);
 Schema schema = new Schema.Parser().parse(new File(schemaFile));
 AvroJob.setInputKeySchema(job, schema);
 AvroJob.setMapOutputKeySchema(job, schema);
 AvroJob.setMapOutputValueSchema(job, schema);
 AvroJob.setOutputKeySchema(job, schema);
 job.setInputFormatClass(AvroKeyInputFormat.class);
 job.setOutputFormatClass(AvroKeyOutputFormat.class);
 job.setOutputKeyClass(AvroKey.class);
 job.setOutputValueClass(NullWritable.class);
 job.setMapperClass(SortMapper.class);
 job.setReducerClass(SortReducer.class);
 return job.waitForCompletion(true) ? 0 : 1;
 }
 
 public static void main(String[] args) throws Exception {
 int exitCode = ToolRunner.run(new AvroSort(), args);
 System.exit(exitCode);
 }
}

정렬은 맵리듀스 셔플 과정에서 일어나며 정렬기능은 에이브로의 스키마에 의해 정해짐

입력데이터 점검

% java -jar $AVRO_HOME/avro-tools-*.jar tojson input/avro/pairs.avro
{"left":"a","right":"1"}
{"left":"c","right":"2"}
{"left":"b","right":"3"}
{"left":"b","right":"2"}

프로그램사용하여 정렬

% hadoop jar avro-examples.jar AvroSort input/avro/pairs.avro output \
 ch12-avro/src/main/resources/SortedStringPair.avsc

정렬 후 저장된 파일 출력

% java -jar $AVRO_HOME/avro-tools-*.jar tojson output/part-r-00000.avro
{"left":"b","right":"3"}
{"left":"b","right":"2"}
{"left":"c","right":"2"}
{"left":"a","right":"1"}