code: @Override
    @SuppressWarnings("unchecked")
    public void putNext(Tuple tuple) throws IOException {
      json.clear();
      if (tuple != null && tuple.size() >= 1) {
        Map<String, Object> map = (Map<String, Object>) tuple.get(0);
        if (keysToKeep_ == null) {
          json.putAll(map);
        } else {
          for (Map.Entry<String, Object> entry : map.entrySet()) {
            if (keysToKeep_.contains(entry.getKey())) {
              json.put(entry.getKey(), entry.getValue());
            }
          }
        }
      }
      try {
        writer.write(null, new Text(json.toString()));
      } catch (InterruptedException e) {
        // Under what circumstances does this happen?
        throw new IOException(e);
      }
    }
docs: The first element is expected to be a map, or null. Anything else causes an error.
  @param tuple the tuple to write.