Расширение TableInputFormat Hadoop для сканирования с префиксом, используемым для распределения ключей меток времени
У меня есть таблица hbase, ключ которой - это временная метка с однобайтовым случайным префиксом для распределения ключей, чтобы при сканировании не возникала горячая точка. Я пытаюсь продлитьTableInputFormat
так что я могу запустить один MapReduce для таблицы с диапазоном, префиксом всех 256 возможных префиксов, чтобы сканировались все диапазоны с указанным диапазоном времени. Однако мое решение не работает, так как оно всегда сканирует последний префикс (127) 256 раз. Что-то должно быть общим для всех сканирований.
Мой код ниже. Есть идеи?
public class PrefixedTableInputFormat extends TableInputFormat {
@Override
public List<InputSplit> getSplits(JobContext context)
throws IOException {
List<InputSplit> splits = new ArrayList<InputSplit>();
Scan scan = getScan();
byte startRow[] = scan.getStartRow(), stopRow[] = scan.getStopRow();
byte prefixedStartRow[] = new byte[startRow.length+1];
byte prefixedStopRow[] = new byte[stopRow.length+1];
System.arraycopy(startRow, 0, prefixedStartRow, 1, startRow.length);
System.arraycopy(stopRow, 0, prefixedStopRow, 1, stopRow.length);
for (int prefix = -128; prefix < 128; prefix++) {
prefixedStartRow[0] = (byte) prefix;
prefixedStopRow[0] = (byte) prefix;
scan.setStartRow(prefixedStartRow);
scan.setStopRow(prefixedStopRow);
setScan(scan);
splits.addAll(super.getSplits(context));
}
return splits;
}
}
а также
Configuration config = HBaseConfiguration.create();
Job job = new Job(config, "Aggregate");
job.setJarByClass(Aggregate.class);
Scan scan = new Scan();
scan.setStartRow("20120630".getBytes());
scan.setStopRow("20120701".getBytes());
scan.setCaching(500);
scan.setCacheBlocks(false);
TableMapReduceUtil.initTableMapperJob(
"event",
scan,
Mapper.class,
ImmutableBytesWritable.class,
ImmutableBytesWritable.class,
job,
true,
PrefixedTableInputFormat.class);
TableMapReduceUtil.initTableReducerJob("event", Reducer.class, job);