import org.apache.pig.data.InternalSortedBag; //导入依赖的package包/类
@Override
public Result getNextTuple() throws ExecException {
	Result res = new Result();

	if (!inputsAccumulated) {
		res = processInput();
           if (!initialized) {
               initialized = true;
               if (PigMapReduce.sJobConfInternal.get() != null) {
                   String bagType = PigMapReduce.sJobConfInternal.get().get(PigConfiguration.PIG_CACHEDBAG_SORT_TYPE);
                   if (bagType != null && bagType.equalsIgnoreCase("default")) {
                       useDefaultBag = true;
                   }
               }
           }
		// by default, we create InternalSortedBag, unless user configures
           // explicitly to use old bag
           sortedBag = useDefaultBag ? BagFactory.getInstance().newSortedBag(mComparator)
                   : new InternalSortedBag(3, mComparator);

           while (res.returnStatus != POStatus.STATUS_EOP) {
			if (res.returnStatus == POStatus.STATUS_ERR) {
				log.error("Error in reading from the inputs");
				return res;
               } else if (res.returnStatus == POStatus.STATUS_NULL) {
                   // Ignore and read the next tuple.
                   res = processInput();
                   continue;
               }
			sortedBag.add((Tuple) res.result);
			res = processInput();
           }

		inputsAccumulated = true;

	}
	if (it == null) {
           it = sortedBag.iterator();
       }
       if (it.hasNext()) {
           res.result = it.next();
           illustratorMarkup(res.result, res.result, 0);
           res.returnStatus = POStatus.STATUS_OK;
       } else {
           res.returnStatus = POStatus.STATUS_EOP;
           reset();
       }
	return res;
}