title: Hadoop mapper key defined as instance variable is hitting performance issue
body: <p>I have files which can have millions of records to process. My processing is segregating records/lines as good or bad. I have implemented it using map reduce.
  There will two types of key, good and bad lines based on the validation of each line.</p>
  
  <p>Here good and bad are keys and each line are values to the respective keys.</p>
  
  <p>Having said so, I can write to context at the end of map task in the below mentioned way:</p>
  
  <pre><code>context.write(new Text("good"), new Text ("line content");
  else{
        context.write(new Text("bad"), new Text ("line content"));
  </code></pre>
  
  <p>On the other hand, I can also declare the keys as instance variables and write to the context. This is taking nearly 15 times the time taken than the first option, for map task to complete for 2 files of 200 MB each.</p>
  
  <p>While I believe, the second option is the better way of programming and should achieve better results. </p>
  
  <p>Could anyone enlighten me what happens when context.write is invoked? since it is the same object, is it taking more time to write i.e. IO??</p>
  
  <p>The second option:</p>
  
  <pre><code>public class ValidatorMapper extends Mapper&lt;LongWritable, Text, Text, Text&gt; {
  
  private Text badKey = new Text();
  private Text goodKey = new Text();
  private Text cleaUpKey = new Text("00");    List&lt;Column&gt; configuredColumns = null;
  
  @Override
  protected void setup(Context context) throws IOException, InterruptedException {
      LOGGER.info("ValidatorMapper : setup");
  
      feedConfig = XMLFeedConfig.getFeed(context.getConfiguration().get(Constants.FEED_CONFIG_STRING));
  .....
  .....
  
  }
  
  @Override
  protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
  
      fileName = ((FileSplit) context.getInputSplit()).getPath().getName();
      goodKey.set(fileName);
      badKey.set("bad-record "+fileName);
      ........
  
  
      System.out.println("inputColumnSeparator len"+inputColumnSeparator);
  
      System.out.println("tokensArray len"+tokensArray.length);
  
      totalRecordNum++;
  
      StringBuilder sb = new StringBuilder();
      for (Column column : configuredColumns) {
          String tokenValue = null;
          if (column.getOutputIndex() &lt;= tokensArray.length &amp;&amp; column.getInputIndex() &lt;= tokensArray.length) {
              tokenValue = tokensArray[column.getInputIndex() - 1];
              if(tokenValue!=null){
                  tokenValue = tokenValue.trim();
              }
              LOGGER.info("value"+ value.toString());
              LOGGER.info("column"+ column.getName());
              LOGGER.info("column"+ column.getInputIndex());
  
  
              LOGGER.info("tokenValue"+ tokenValue);
              LOGGER.info("tokenValue"+ tokensArray.toString());
  
  
              validatedMap = DataTypeValidator.validate(column, tokenValue);
              errorMesg = validatedMap.get(Constants.ERROR_MSG);
              newDateValue = validatedMap.get(Constants.NEW_DATE);
              if (null == errorMesg &amp;&amp; null != newDateValue) {
                  tokenValue = newDateValue;
              }
          } else {
              if(Constants.DELIMETER.equalsIgnoreCase(dataSeperator))
              {
                  errorMesg = Constants.COLUMN_MISMATCH;
  
              }
          }
          if (errorMesg != null) {
              badFeedCount++;
              context.write(badKey, new Text(value + "bad-line no:" + key));
              break;
          } else {
              sb.append(tokenValue + outputColumnSeparator);
          }
      }
      LOGGER.info("errorMesg = " + errorMesg);
      if (errorMesg == null) {
          String processedValue = sb.substring(0, sb.lastIndexOf(outputColumnSeparator));
          processedRecord.set(processedValue);
          context.write(goodKey, processedRecord);
      }
  }
  </code></pre>