D-Net

    private Logger log = Logger.getLogger(this.getClass());

    private Serializer serializer;

    private EntityConfigTable entityConfigTable;

    private long threshold = 100000;

    @Override

    protected void setup(Context context) throws IOException, InterruptedException {

        loadEntityConfig(context);

        serializer = new Serializer(context.getConfiguration().get("lod.delim"), context.getConfiguration().get("lod.enclosing"));

    }

    @Override

    protected void map(final ImmutableBytesWritable keyIn, final Result result, final Context context) throws IOException {

        final OafRowKeyDecoder decoder = OafRowKeyDecoder.decode(keyIn.copyBytes());

        final Type type = decoder.getType();

        final Oaf oaf = UpdateMerger.mergeBodyUpdates(context, result.getFamilyMap(Bytes.toBytes(type.toString())));

        if (isValid(oaf)) {

            // TODO set this only when the configuration file has include dups

            // to fals for Results

            // or else were gonna get deleted by inference entries

            // if (!deletedByInference(oaf) ||

            // entityConfigTable.includeDuplicates(type)) {

            if (!deletedByInference(oaf)) {

                emitProtos(context, result, oaf);

            }

        }

    }

    private boolean isValid(Oaf oaf) {

        try {

            if (oaf != null && oaf.isInitialized()) {

                return true;

            }

        } catch (Exception e) {

            // log.error("OAF NOT INITIALIZED ");

        }

        return false;

    }

    private void emitProtos(Context context, Result result, Oaf oaf) {

        //Oaf.Builder oafBuilder = Oaf.newBuilder().setKind(oaf.getKind()).setDataInfo(oaf.getDataInfo()).setTimestamp(oaf.getTimestamp());

        Oaf.Builder oafBuilder = Oaf.newBuilder().setKind(oaf.getKind()).setDataInfo(oaf.getDataInfo());

        Type type = oaf.getEntity().getType();

        oafBuilder.setEntity(oaf.getEntity());

        // emit relation first so we can cache them to entity protos

        emitRelation(context, result, oaf, type, oafBuilder);

        emitEntity(context, oaf, type, oafBuilder);

    }

    private void emitEntity(Context context, Oaf oaf, Type type, Oaf.Builder oafBuilder) {

        String serialized = serializer.serialize(oafBuilder.build());

        if (serialized != null && !oaf.getEntity().getId().startsWith("dedup")) {

            try {

                Text TextKeyOut = new Text("entities");

                context.write((TextKeyOut), new ImmutableBytesWritable(serialized.getBytes()));

            } catch (IOException e) {

                log.error("Error writing entity to M/R output", e);

            } catch (InterruptedException e) {

                log.error("Error writing entity to M/R output: Job Interrupted.", e);

            }

        }

    }

    // may have multiple relations per each field

    private void emitRelation(Context context, Result result, Oaf oaf, Type type, Oaf.Builder builder) {

        // derived relations; ResultLanguages, resultConcepts etc are

        // created here

        if (!oaf.getEntity().getId().startsWith("dedup")) {

            // Existing Hbase relations are generated here

            if (entityConfigTable.getDescriptors(type) != null && !entityConfigTable.getDescriptors(type).isEmpty()) {

                for (LinkDescriptor ld : entityConfigTable.getDescriptors(type)) {

                    try {

                        final Map<byte[], byte[]> columnMap = result.getFamilyMap(Bytes.toBytes(ld.getRelDescriptor().getIt()));

                        ArrayList<OafRel> relOaf = decodeRelation(oaf, context, columnMap, ld);

                        for (OafRel rel : relOaf) {

                            builder.getEntityBuilder().addCachedRel(rel);

                            try {

                                Text TextKeyOut = new Text("relations");

                                String buff = serializer.serialize(rel);

                                context.write((TextKeyOut), new ImmutableBytesWritable(buff.getBytes()));

                            } catch (Exception e) {

                                log.error("Error while writing Relation Proto to M/R output", e);

                            }

                        }

                    } catch (Exception e) {

                        log.error("Error while decoding Relation Proto from HBase Body", e);

                    }

                }

            }

        }

        List<String> relationsList = serializer.extractRelations(oaf);

        for (String rel : relationsList) {

            try {

                Text TextKeyOut = new Text("relations");

                if (oaf.getEntity().getId().startsWith("dedup")) {

                    if (rel.startsWith("Dedup")) {

                        context.write((TextKeyOut), new ImmutableBytesWritable(rel.getBytes()));

                    }

                } else {

                    context.write((TextKeyOut), new ImmutableBytesWritable(rel.getBytes()));

                }

            } catch (Exception e) {

                log.error("Error writing relations to output : " + rel);

            }

        }

    }

    private ArrayList<OafRel> decodeRelation(final Oaf body, final Context context, Map<byte[], byte[]> columnMap, final LinkDescriptor ld) throws IOException, InterruptedException {

        ArrayList<OafRel> rels = new ArrayList<OafRel>();

        if (hasData(columnMap)) {

            for (Entry<byte[], byte[]> e : columnMap.entrySet()) {

                final Oaf decodedOaf = decodeProto(context, e.getValue());

                if (isValid(decodedOaf) && !deletedByInference(decodedOaf)) {

                    OafRel.Builder relBuilder = OafRel.newBuilder(decodedOaf.getRel());

                    // skip dedups

                    if (ld.getRelDescriptor().getIt().contains(SubRelType.dedup.toString()) && isDedupSelf(relBuilder)) {

                        //		OafRel oafRel = relBuilder.setRelType(ld.getRelDescriptor().getRelType()).build();

                        //		relBuilder.setCachedTarget(body.getEntity());

                        //	rels.put("dedups", oafRel);

                    } else {

                        OafRel oafRel = relBuilder.setRelType(ld.getRelDescriptor().getRelType()).build();

                        relBuilder.setCachedTarget(body.getEntity());

                        rels.add(oafRel);

                    }

                }

            }

        }

        return rels;

    }

    private Oaf decodeProto(final Context context, final byte[] body) {

        try {

            return Oaf.parseFrom(body);

        } catch (InvalidProtocolBufferException e) {

            // log.error("Corrupted proto ", e);

        }

        return null;

    }

    private void loadEntityConfig(Context context) {

        String indexConf = context.getConfiguration().get("index.conf");

        if (indexConf == null || indexConf.isEmpty()) {

            log.error("NULL ENTITY CONFIGURATION TABLE IN MAPPER  : ");

        }

        entityConfigTable = IndexConfig.load(indexConf).getConfigMap();

    }

    private boolean isDedupSelf(final OafRelOrBuilder rel) {

        return rel.getSource().contains(rel.getTarget());

    }

    private boolean hasData(final Map<byte[], byte[]> columnMap) {

        return columnMap != null && !columnMap.isEmpty();

    }

    private boolean deletedByInference(final Oaf oaf) {

        return oaf.getDataInfo().getDeletedbyinference();

    }

    @Override

    protected void cleanup(Context context) throws IOException, InterruptedException {

        super.cleanup(context);

    }

    public Serializer getSerializer() {

        return serializer;

    }

    public void setSerializer(Serializer serializer) {

        this.serializer = serializer;

    }

    public EntityConfigTable getEntityConfigTable() {

        return entityConfigTable;

    }

    public void setEntityConfigTable(EntityConfigTable entityConfigTable) {

        this.entityConfigTable = entityConfigTable;

    }

    public long getThreshold() {

        return threshold;

    }

    public void setThreshold(long threshold) {

        this.threshold = threshold;

    }