| 1 |
53588
|
sandro.lab
|
package eu.dnetlib.data.mapreduce.hbase.dataexport;
|
| 2 |
|
|
|
| 3 |
53604
|
claudio.at
|
import com.google.common.base.Splitter;
|
| 4 |
53628
|
claudio.at
|
import com.google.common.collect.Lists;
|
| 5 |
53604
|
claudio.at
|
import com.google.common.collect.Sets;
|
| 6 |
53588
|
sandro.lab
|
import com.googlecode.protobuf.format.JsonFormat;
|
| 7 |
|
|
import eu.dnetlib.data.mapreduce.util.DedupUtils;
|
| 8 |
|
|
import eu.dnetlib.data.proto.OafProtos;
|
| 9 |
|
|
import eu.dnetlib.data.proto.TypeProtos;
|
| 10 |
|
|
import org.apache.hadoop.hbase.client.Result;
|
| 11 |
|
|
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
|
| 12 |
|
|
import org.apache.hadoop.hbase.mapreduce.TableMapper;
|
| 13 |
|
|
import org.apache.hadoop.io.Text;
|
| 14 |
53604
|
claudio.at
|
import org.apache.hadoop.mapreduce.Mapper;
|
| 15 |
53588
|
sandro.lab
|
|
| 16 |
|
|
import java.io.IOException;
|
| 17 |
|
|
import java.util.Map;
|
| 18 |
53604
|
claudio.at
|
import java.util.Set;
|
| 19 |
53588
|
sandro.lab
|
|
| 20 |
53604
|
claudio.at
|
public class ConfigurableExportMapper extends TableMapper<Text, Text> {
|
| 21 |
53588
|
sandro.lab
|
|
| 22 |
53604
|
claudio.at
|
private static final String CF_DEFAULT = "result";
|
| 23 |
|
|
private static final String QUALIFIER_DEFAULT = "body";
|
| 24 |
|
|
|
| 25 |
53588
|
sandro.lab
|
private Text keyOut;
|
| 26 |
|
|
private Text valueOut;
|
| 27 |
|
|
|
| 28 |
53604
|
claudio.at
|
private Set<String> columnFamilies = Sets.newHashSet();
|
| 29 |
|
|
private Set<String> qualifiers = Sets.newHashSet();
|
| 30 |
|
|
|
| 31 |
|
|
enum ExportMode {
|
| 32 |
|
|
ALL,
|
| 33 |
|
|
DUPS_ONLY
|
| 34 |
|
|
}
|
| 35 |
|
|
|
| 36 |
|
|
private ExportMode exportMode = ExportMode.ALL;
|
| 37 |
|
|
|
| 38 |
53588
|
sandro.lab
|
@Override
|
| 39 |
|
|
protected void setup(Context context) throws IOException, InterruptedException {
|
| 40 |
|
|
keyOut = new Text("");
|
| 41 |
|
|
valueOut = new Text();
|
| 42 |
53604
|
claudio.at
|
|
| 43 |
|
|
|
| 44 |
53628
|
claudio.at
|
columnFamilies.addAll(Lists.newArrayList(Splitter.on(",").split(context.getConfiguration().get("column.families", CF_DEFAULT))));
|
| 45 |
|
|
qualifiers.addAll(Lists.newArrayList(Splitter.on(",").split(context.getConfiguration().get("qualifiers", QUALIFIER_DEFAULT))));
|
| 46 |
53604
|
claudio.at
|
exportMode = ExportMode.valueOf(context.getConfiguration().get("export.mode", ExportMode.ALL.toString()));
|
| 47 |
|
|
|
| 48 |
53588
|
sandro.lab
|
}
|
| 49 |
|
|
|
| 50 |
|
|
@Override
|
| 51 |
|
|
protected void map(ImmutableBytesWritable key, Result value, Context context) throws IOException, InterruptedException {
|
| 52 |
53604
|
claudio.at
|
for(String cf : columnFamilies) {
|
| 53 |
|
|
switch (exportMode) {
|
| 54 |
|
|
case ALL:
|
| 55 |
|
|
export(value, context, cf);
|
| 56 |
|
|
break;
|
| 57 |
|
|
case DUPS_ONLY:
|
| 58 |
|
|
final Map<byte[], byte[]> mergedInMap = value.getFamilyMap(DedupUtils.getDedupCF_mergedInBytes(cf));
|
| 59 |
|
|
if (mergedInMap != null && !mergedInMap.isEmpty()) {
|
| 60 |
|
|
export(value, context, cf);
|
| 61 |
|
|
}
|
| 62 |
53588
|
sandro.lab
|
|
| 63 |
53604
|
claudio.at
|
break;
|
| 64 |
|
|
}
|
| 65 |
|
|
}
|
| 66 |
|
|
}
|
| 67 |
53588
|
sandro.lab
|
|
| 68 |
53604
|
claudio.at
|
private void export(Result value, Context context, String cf) throws IOException, InterruptedException {
|
| 69 |
|
|
for(String q : qualifiers) {
|
| 70 |
53588
|
sandro.lab
|
|
| 71 |
53604
|
claudio.at
|
final byte[] data = value.getValue(cf.getBytes(), q.getBytes());
|
| 72 |
|
|
|
| 73 |
|
|
if (data != null) {
|
| 74 |
|
|
OafProtos.Oaf oaf = OafProtos.Oaf.parseFrom(data);
|
| 75 |
53588
|
sandro.lab
|
valueOut.set(JsonFormat.printToString(oaf));
|
| 76 |
|
|
context.write(keyOut, valueOut);
|
| 77 |
53604
|
claudio.at
|
|
| 78 |
|
|
context.getCounter(cf, q).increment(1);
|
| 79 |
53588
|
sandro.lab
|
}
|
| 80 |
|
|
}
|
| 81 |
53604
|
claudio.at
|
}
|
| 82 |
53588
|
sandro.lab
|
|
| 83 |
53604
|
claudio.at
|
|
| 84 |
53588
|
sandro.lab
|
}
|