Hbase MapReduce: как использовать пользовательский класс в качестве значения для преобразователя и/или редуктора?
Я пытаюсь ознакомиться с заданиями Hadoop/Hbase MapReduce, чтобы иметь возможность правильно их писать. Прямо сейчас у меня есть экземпляр Hbase с таблицей dns с некоторыми записями DNS. Я попытался сделать простой счетчик уникальных доменов, который выводит файл, и это сработало. Прямо сейчас я использую только IntWritable или Text, и мне было интересно, можно ли использовать пользовательские объекты для моего Mapper/Reducer. Я пытался сделать это сам, но я получаю
Error: java.io.IOException: Initialization of all the collectors failed. Error in last collector was :null
at org.apache.hadoop.mapred.MapTask.createSortingCollector(MapTask.java:415)
at org.apache.hadoop.mapred.MapTask.access$100(MapTask.java:81)
at org.apache.hadoop.mapred.MapTask$NewOutputCollector.<init>(MapTask.java:698)
at org.apache.hadoop.mapred.MapTask.runNewMapper(MapTask.java:770)
at org.apache.hadoop.mapred.MapTask.run(MapTask.java:341)
at org.apache.hadoop.mapred.YarnChild$2.run(YarnChild.java:170)
at java.security.AccessController.doPrivileged(Native Method)
at javax.security.auth.Subject.doAs(Subject.java:422)
at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1869)
at org.apache.hadoop.mapred.YarnChild.main(YarnChild.java:164)
Caused by: java.lang.NullPointerException
at org.apache.hadoop.mapred.MapTask$MapOutputBuffer.init(MapTask.java:1011)
at org.apache.hadoop.mapred.MapTask.createSortingCollector(MapTask.java:402)
... 9 more
Так как я новичок в этом, я на самом деле не знаю, что делать. Я предполагаю, что мне нужно реализовать один или несколько интерфейсов или расширить абстрактный класс, но я не могу найти здесь или в Интернете подходящий пример.
Я попытался сделать простой счетчик доменов из моей таблицы DNS, но используя класс в качестве оболочки над целым числом (только в дидактических целях). Мой Класс карты выглядит так:
public class Map extends TableMapper<Text, MapperOutputValue> {
private static byte[] columnName = "fqdn".getBytes();
private static byte[] columnFamily = "d".getBytes();
public void map(ImmutableBytesWritable row, Result value, Context context)
throws InterruptedException, IOException {
String fqdn = new String(value.getValue(columnFamily, columnName));
Text key = new Text();
key.set(fqdn);
context.write(key, new MapperOutputValue(1));
}
}
Редуктор:
public class Reduce extends Reducer<Text, MapperOutputValue, Text, IntWritable> {
@Override
public void reduce(Text key, Iterable<MapperOutputValue> values, Context context)
throws IOException, InterruptedException {
int i = 0;
for (MapperOutputValue val : values) {
i += val.getCount();
}
context.write(key, new IntWritable(i));
}
}
И часть моей функции Водитель/Главный:
TableMapReduceUtil.initTableMapperJob(
"dns",
scan,
Map.class,
Text.class,
MapperOutputValue.class,
job);
/* Set output parameters */
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
job.setOutputFormatClass(TextOutputFormat.class);
Как я уже сказал, MapperOutputValue — это просто простой класс, который содержит приватное целое число, конструктор с параметром, геттер и сеттер. Я также пытался добавить метод toString, но он все еще не работает.
Итак, мой вопрос: как лучше всего использовать пользовательские классы в качестве вывода картографа/ввода для редуктора? Кроме того, допустим, я хочу использовать класс с несколькими полями в качестве конечного вывода редуктора. Что этот класс должен реализовать/расширить? Это хорошая идея, или я должен придерживаться использования «примитивов» в качестве IntWritable или Text?
Благодарить!
@BenWatson после того, как я опубликовал это, я реализовал интерфейс WritableComparable и смог заставить его работать, но только с целыми числами. Я не знаю, какие методы лучше всего подходят для работы со строками. В любом случае спасибо за статью
@BenWatson, вы можете ответить примером из этой статьи (или другой, как хотите), я приму это. Я заставил его работать так, как я хочу, статья была очень полезной. Спасибо!
Рад, что смог помочь.
Ответы 1
MapOutputValue должен реализовывать Writable, чтобы его можно было сериализовать между задачами в задании MapReduce. Замена MapOutputJob на следующее должно работать:
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
public class DomainCountWritable implements Writable {
private Text domain;
private IntWritable count;
public DomainCountWritable() {
this.domain = new Text();
this.count = new IntWritable(0);
}
public DomainCountWritable(Text domain, IntWritable count) {
this.domain = domain;
this.count = count;
}
public Text getDomain() {
return this.domain;
}
public IntWritable getCount() {
return this.count;
}
public void setDomain(Text domain) {
this.domain = domain;
}
public void setCount(IntWritable count) {
this.count = count;
}
public void readFields(DataInput in) throws IOException {
this.domain.readFields(in);
this.count.readFields(in);
}
public void write(DataOutput out) throws IOException {
this.domain.write(out);
this.count.write(out);
}
@Override
public String toString() {
return this.domain.toString() + "\t" + this.count.toString();
}
}
MapperOutputValueреализуетWritable? bigdatums.net/2016/06/05/…. Если вы не так хорошо знакомы с MapReduce, возможно, стоит разбить это на более простую задачу, где вы читаете записи DNS из файла HDFS, а затем добавляете подключение HBase, как только это сработает.