@Override
  protected void reduce(ImmutableBytesWritable rowKey, java.lang.Iterable<Text> lines,
    Reducer<ImmutableBytesWritable, Text, ImmutableBytesWritable, KeyValue>.Context context)
    throws java.io.IOException, InterruptedException {
    // although reduce() is called per-row, handle pathological case
    long threshold = context.getConfiguration().getLong("reducer.row.threshold", 1L * (1 << 30));
    Iterator<Text> iter = lines.iterator();
    while (iter.hasNext()) {
      Set<KeyValue> kvs = new TreeSet<>(CellComparator.getInstance());
      long curSize = 0;
      // stop at the end or the RAM threshold
      while (iter.hasNext() && curSize < threshold) {
        Text line = iter.next();
        byte[] lineBytes = line.getBytes();
        try {
          ImportTsv.TsvParser.ParsedLine parsed = parser.parse(lineBytes, line.getLength());
          // Retrieve timestamp if exists
          ts = parsed.getTimestamp(ts);
          cellVisibilityExpr = parsed.getCellVisibility();
          ttl = parsed.getCellTTL();

          // create tags for the parsed line
          List<Tag> tags = new ArrayList<>();
          if (cellVisibilityExpr != null) {
            tags.addAll(kvCreator.getVisibilityExpressionResolver()
              .createVisibilityExpTags(cellVisibilityExpr));
          }
          // Add TTL directly to the KV so we can vary them when packing more than one KV
          // into puts
          if (ttl > 0) {
            tags.add(new ArrayBackedTag(TagType.TTL_TAG_TYPE, Bytes.toBytes(ttl)));
          }
          for (int i = 0; i < parsed.getColumnCount(); i++) {
            if (
              i == parser.getRowKeyColumnIndex() || i == parser.getTimestampKeyColumnIndex()
                || i == parser.getAttributesKeyColumnIndex()
                || i == parser.getCellVisibilityColumnIndex() || i == parser.getCellTTLColumnIndex()
            ) {
              continue;
            }
            // Creating the KV which needs to be directly written to HFiles. Using the Facade
            // KVCreator for creation of kvs.
            ExtendedCell cell = (ExtendedCell) this.kvCreator.create(lineBytes,
              parsed.getRowKeyOffset(), parsed.getRowKeyLength(), parser.getFamily(i), 0,
              parser.getFamily(i).length, parser.getQualifier(i), 0, parser.getQualifier(i).length,
              ts, lineBytes, parsed.getColumnOffset(i), parsed.getColumnLength(i), tags);
            KeyValue kv = KeyValueUtil.ensureKeyValue(cell);
            kvs.add(kv);
            curSize += kv.heapSize();
          }
        } catch (ImportTsv.TsvParser.BadTsvLineException | IllegalArgumentException
          | InvalidLabelException badLine) {
          if (skipBadLines) {
            System.err.println("Bad line." + badLine.getMessage());
            incrementBadLineCount(1);
            continue;
          }
          throw new IOException(badLine);
        }
      }
      context.setStatus("Read " + kvs.size() + " entries of " + kvs.getClass() + "("
        + Strings.humanReadableInt(curSize) + ")");
      int index = 0;
      for (KeyValue kv : kvs) {
        context.write(rowKey, kv);
        if (++index > 0 && index % 100 == 0) context.setStatus("Wrote " + index + " key values.");
      }

      // if we have more entries to process
      if (iter.hasNext()) {
        // force flush because we cannot guarantee intra-row sorted order
        context.write(null, null);
      }
    }
  }
Handles common parameter initialization that a subclass might want to leverage.
hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/TextSortReducer.java