@Override
      public void reduce(BytesWritable key, Iterable<BytesWritable> values, Context context)
          throws IOException, InterruptedException {
        int defCount = 0;
        refs.clear();
        // We only expect two values, a DEF and a reference, but there might be more.
        for (BytesWritable type : values) {
          if (type.getLength() == DEF.getLength()) {
            defCount++;
          } else {
            byte[] bytes = new byte[type.getLength()];
            System.arraycopy(type.getBytes(), 0, bytes, 0, type.getLength());
            refs.add(bytes);
          }
        }
        // TODO check for more than one def, should not happen
        List<String> refsList = new ArrayList<>(refs.size());
        String keyString = null;
        if (defCount == 0 || refs.size() != 1) {
          for (byte[] ref : refs) {
            refsList.add(COMMA_JOINER.join(Bytes.getLong(ref), Bytes.getLong(ref, 8)));
          }
          keyString = COMMA_JOINER.join(Bytes.getLong(key.getBytes()),
              Bytes.getLong(key.getBytes(), 8));
          LOG.error("Linked List error: Key = " + keyString + " References = " + refsList);
        }
        if (defCount == 0 && refs.size() > 0) {
          // this is bad, found a node that is referenced but not defined. It must have been
          // lost, emit some info about this node for debugging purposes.
          context.write(new Text(keyString), new Text(refsList.toString()));
          context.getCounter(Counts.UNDEFINED).increment(1);
        } else if (defCount > 0 && refs.size() == 0) {
          // node is defined but not referenced
          context.write(new Text(keyString), new Text("none"));
          context.getCounter(Counts.UNREFERENCED).increment(1);
        } else {
          if (refs.size() > 1) {
            if (refsList != null) {
              context.write(new Text(keyString), new Text(refsList.toString()));
            }
            context.getCounter(Counts.EXTRAREFERENCES).increment(refs.size() - 1);
          }
          // node is defined and referenced
          context.getCounter(Counts.REFERENCED).increment(1);
        }
      }
@Override
      public void reduce(BytesWritable key, Iterable<BytesWritable> values, Context context)
          throws IOException, InterruptedException {
        int defCount = 0;
        refs.clear();
       
        for (BytesWritable type : values) {
          if (type.getLength() == DEF.getLength()) {
            defCount++;
          } else {
            byte[] bytes = new byte[type.getLength()];
            System.arraycopy(type.getBytes(), 0, bytes, 0, type.getLength());
            refs.add(bytes);
          }
        }
       
        List<String> refsList = new ArrayList<>(refs.size());
        String keyString = null;
        if (defCount == 0 || refs.size() != 1) {
          for (byte[] ref : refs) {
            refsList.add(COMMA_JOINER.join(Bytes.getLong(ref), Bytes.getLong(ref, 8)));
          }
          keyString = COMMA_JOINER.join(Bytes.getLong(key.getBytes()),
              Bytes.getLong(key.getBytes(), 8));
          LOG.error("Linked List error: Key = " + keyString + " References = " + refsList);
        }
        if (defCount == 0 && refs.size() > 0) {
         
         
          context.write(new Text(keyString), new Text(refsList.toString()));
          context.getCounter(Counts.UNDEFINED).increment(1);
        } else if (defCount > 0 && refs.size() == 0) {
         
          context.write(new Text(keyString), new Text("none"));
          context.getCounter(Counts.UNREFERENCED).increment(1);
        } else {
          if (refs.size() > 1) {
            if (refsList != null) {
              context.write(new Text(keyString), new Text(refsList.toString()));
            }
            context.getCounter(Counts.EXTRAREFERENCES).increment(refs.size() - 1);
          }
         
          context.getCounter(Counts.REFERENCED).increment(1);
        }
      }
@override public void reduce(byteswritable key, iterable<byteswritable> values, context context) throws ioexception, interruptedexception { int defcount = 0; refs.clear(); for (byteswritable type : values) { if (type.getlength() == def.getlength()) { defcount++; } else { byte[] bytes = new byte[type.getlength()]; system.arraycopy(type.getbytes(), 0, bytes, 0, type.getlength()); refs.add(bytes); } } list<string> refslist = new arraylist<>(refs.size()); string keystring = null; if (defcount == 0 || refs.size() != 1) { for (byte[] ref : refs) { refslist.add(comma_joiner.join(bytes.getlong(ref), bytes.getlong(ref, 8))); } keystring = comma_joiner.join(bytes.getlong(key.getbytes()), bytes.getlong(key.getbytes(), 8)); log.error("linked list error: key = " + keystring + " references = " + refslist); } if (defcount == 0 && refs.size() > 0) { context.write(new text(keystring), new text(refslist.tostring())); context.getcounter(counts.undefined).increment(1); } else if (defcount > 0 && refs.size() == 0) { context.write(new text(keystring), new text("none")); context.getcounter(counts.unreferenced).increment(1); } else { if (refs.size() > 1) { if (refslist != null) { context.write(new text(keystring), new text(refslist.tostring())); } context.getcounter(counts.extrareferences).increment(refs.size() - 1); } context.getcounter(counts.referenced).increment(1); } }