pig/src/main/java/com/twitter/elephantbird/pig/load/JsonLoader.java
@Override
  public Tuple getNext() throws IOException {
    if (reader == null) {
      return null;
    }
    // nested load can be enabled through a pig property
    if ("true".equals(jobConf.get(NESTED_LOAD_KEY)))
      isNestedLoadEnabled = true;
    try {
      while (reader.nextKeyValue()) {
        Text value = (Text) reader.getCurrentValue();
        incrCounter(JsonLoaderCounters.LinesRead, 1L);
        Tuple t = parseStringToTuple(value.toString());
        if (t != null) {
          incrCounter(JsonLoaderCounters.LinesJsonDecoded, 1L);
          return t;
        }
      }
      return null;

    } catch (InterruptedException e) {
      int errCode = 6018;
      String errMsg = "Error while reading input";
      throw new ExecException(errMsg, errCode,
          PigException.REMOTE_ENVIRONMENT, e);
    }
  }
@Override
  public Tuple getNext() throws IOException {
    if (reader == null) {
      return null;
    }
    // nested load can be enabled through a pig property
    if ("true".equals(jobConf.get(NESTED_LOAD_KEY)))
      isNestedLoadEnabled = true;
    try {
      while (reader.nextKeyValue()) {
        Text value = (Text) reader.getCurrentValue();
        incrCounter(JsonLoaderCounters.LinesRead, 1L);
        Tuple t = parseStringToTuple(value.toString());
        if (t != null) {
          incrCounter(JsonLoaderCounters.LinesJsonDecoded, 1L);
          return t;
        }
      }
      return null;

    } catch (InterruptedException e) {
      int errCode = 6018;
      String errMsg = "Error while reading input";
      throw new ExecException(errMsg, errCode,
          PigException.REMOTE_ENVIRONMENT, e);
    }
  }
Return every non-null line as a single-element tuple to Pig.