org.apache.hadoop.io.LongWritable#set ( )源码实例Demo

@Override
public synchronized boolean next(LongWritable key, Text value)
        throws IOException
{
    while (true) {
        int bytes = readLine(value);
        if (bytes <= 0) {
            if (!selectClient.isRequestComplete()) {
                throw new IOException("S3 Select request was incomplete as End Event was not received");
            }
            return false;
        }
        recordsFromS3++;
        if (recordsFromS3 > processedRecords) {
            position += bytes;
            processedRecords++;
            key.set(processedRecords);
            return true;
        }
    }
}
 

private void writeInStoreHour() throws IOException, InterruptedException{

        KeyWrapper cycleKey = new KeyWrapper();
        cycleKey.setType(new Text(MapKeyConfig.IN_STORE_HOUR));

        LongWritable longWritable = new LongWritable();
        cycleKey.setMillisTime(longWritable);

        IntWritable value = new IntWritable(1);


        List<Long> inStoreHours = statistic.getInStoreHours();
        for (Long inStoreTime : inStoreHours) {
            longWritable.set(IntervalCalculator.getInStoreInterval(inStoreTime));
            context.write(cycleKey, new ValueWrapper(value));
        }
    }
 

public void reduce(LongWritable key, Iterator<Text> values,
                   OutputCollector<LongWritable, Text> output,
                   Reporter reporter) throws IOException {
  while (values.hasNext()) {
    Text value = values.next();
    writeFlag(conf, "reduce." + name + ".value." + value);
    key.set(10);
    output.collect(key, value);
    if (byValue) {
      assertEquals(10, key.get());
    } else {
      assertNotSame(10, key.get());
    }
    key.set(11);
  }
}
 

/** {@inheritDoc} */
public boolean next(LongWritable key, T value) throws IOException {
  try {
    if (!results.next())
      return false;

    // Set the key field value as the output key value
    key.set(pos + split.getStart());

    value.readFields(results);

    pos ++;
  } catch (SQLException e) {
    throw new IOException(e.getMessage());
  }
  return true;
}
 

/** Read a line. */
public  boolean next(LongWritable key, Text value)
throws IOException {
    if (pos > end)
        return false;

    key.set(pos); // key is position
    buffer.reset();
    // long bytesRead = LineRecordReader.readLine(in, buffer); 
    long bytesRead = readLine(in, buffer);
    if (bytesRead == 0) {
        return false;
    }
    pos = in.getPos();
    // if we have read ahead because we encountered a carriage return
    // char followed by a non line feed char, decrement the pos
    if(CRFollowedByNonLF) {
        pos--;
    }

    bridge.target = value;
    buffer.writeTo(bridge);
    return true;
}
 

@Override
public void postSuperstep() {
    super.postSuperstep();

    try {
        for (Map.Entry<String, AggregationStorage> aggregationStorageEntry : aggregationStorages.entrySet()) {
            String aggregationStorageName = aggregationStorageEntry.getKey();
            AggregationStorage aggregationStorage = aggregationStorageEntry.getValue();

            workerContext.addAggregationStorage(aggregationStorageName, aggregationStorage);
        }
    } catch (RuntimeException e) {
        LOG.error(e);
        throw e;
    }

    LongWritable longWritable = new LongWritable();

    LOG.info("Num embeddings processed: " + numEmbeddingsProcessed);
    longWritable.set(numEmbeddingsProcessed);
    aggregate(MasterExecutionEngine.AGG_EMBEDDINGS_PROCESSED, longWritable);
    LOG.info("Num embeddings generated: " + numEmbeddingsGenerated);
    longWritable.set(numEmbeddingsGenerated);
    aggregate(MasterExecutionEngine.AGG_EMBEDDINGS_GENERATED, longWritable);
    LOG.info("Num embeddings output: " + numberOfEmbeddingsOutput);
    longWritable.set(numberOfEmbeddingsOutput);
    aggregate(MasterExecutionEngine.AGG_EMBEDDINGS_OUTPUT, longWritable);
}
 

@Override
public void finish() {
    flush();

    LongWritable longWritable = new LongWritable();
    longWritable.set(totalSizeODAGs);

    getExecutionEngine().aggregate(MasterExecutionEngine.AGG_PROCESSED_SIZE_ODAG, longWritable);
}
 

/** Read a line. */
public synchronized boolean next(LongWritable key, Text value)
		throws IOException {
	while (pos < end) {
		key.set(pos);

		int newSize = lineReader.readLine(value, maxLineLength,
				Math.max((int) Math.min(Integer.MAX_VALUE, end - pos),
						maxLineLength));

		// start
		String strReplace = value.toString().toLowerCase()
				.replaceAll("\\|\\|\\|", "\001");
		Text txtReplace = new Text();
		txtReplace.set(strReplace);
		value.set(txtReplace.getBytes(), 0, txtReplace.getLength());
		// end

		if (newSize == 0) {
			return false;
		}
		pos += newSize;
		if (newSize < maxLineLength) {
			return true;
		}

		// line too long. try again
		LOG.info("Skipped line of size " + newSize + " at pos "
				+ (pos - newSize));
	}

	return false;
}
 

@Override
public synchronized boolean next(LongWritable key, BytesWritable value)
    throws IOException {
  boolean dataRead = reader.nextKeyValue();
  if (dataRead) {
    LongWritable newKey = reader.getCurrentKey();
    BytesWritable newValue = reader.getCurrentValue();
    key.set(newKey.get());
    value.set(newValue);
  }
  return dataRead;
}
 

@Override
public synchronized boolean next(LongWritable key, BytesWritable value)
		throws IOException {
	boolean dataRead = reader.nextKeyValue();
	if (dataRead) {
		LongWritable newKey = reader.getCurrentKey();
		BytesWritable newValue = reader.getCurrentValue();
		key.set(newKey.get());
		value.set(newValue);
	}
	return dataRead;
}
 

private void writCustomerFlow() throws IOException, InterruptedException{

        KeyWrapper customerFlowKey = new KeyWrapper();
        customerFlowKey.setType(new Text(MapKeyConfig.CUSTOMER_FLOW_KEY));

        LongWritable longWritable = new LongWritable();
        customerFlowKey.setMillisTime(longWritable);

        for (CustomerFlowElement customerFlowElement:statistic.getCustomerFlowElements()) {
            longWritable.set(customerFlowElement.getHour());
            context.write(customerFlowKey, new ValueWrapper(customerFlowElement));
        }
    }
 

private void writeCycle() throws IOException, InterruptedException{

        KeyWrapper cycleKey = new KeyWrapper();
        cycleKey.setType(new Text(MapKeyConfig.CYCLE));

        LongWritable longWritable = new LongWritable();
        cycleKey.setMillisTime(longWritable);

        IntWritable value = new IntWritable(1);

        for (Long cycle : statistic.getCycles()) {
            longWritable.set(IntervalCalculator.getCycleInterval(cycle));
            context.write(cycleKey, new ValueWrapper(value));
        }
    }
 

/** Read a line. */
public synchronized boolean next(LongWritable key, Text value)
  throws IOException {

  // We always read one extra line, which lies outside the upper
  // split limit i.e. (end - 1)
  while (getFilePosition() <= end || in.needAdditionalRecordAfterSplit()) {
    key.set(pos);

    int newSize = 0;
    if (pos == 0) {
      newSize = skipUtfByteOrderMark(value);
    } else {
      newSize = in.readLine(value, maxLineLength, maxBytesToConsume(pos));
      pos += newSize;
    }

    if (newSize == 0) {
      return false;
    }
    if (newSize < maxLineLength) {
      return true;
    }

    // line too long. try again
    LOG.info("Skipped line of size " + newSize + " at pos " + (pos - newSize));
  }

  return false;
}
 

@Override
public synchronized boolean next(LongWritable key, BytesWritable value)
    throws IOException {
  boolean dataRead = reader.nextKeyValue();
  if (dataRead) {
    LongWritable newKey = reader.getCurrentKey();
    BytesWritable newValue = reader.getCurrentValue();
    key.set(newKey.get());
    value.set(newValue);
  }
  return dataRead;
}
 

public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    VarLongWritable userID = new VarLongWritable();
    LongWritable itemID = new LongWritable();
    FloatWritable itemValue = new FloatWritable();System.err.println("key:"+key+"    value:"+value+"   ");
    String line = value.toString();
    String[] info = line.split(",");
    if (info.length != 3) {
        return;
    }
    userID.set(Long.parseLong(info[0]));
    itemID.set(Long.parseLong(info[1]));
    itemValue.set(Float.parseFloat(info[2]));
    context.write(userID, new LongAndFloat(itemID, itemValue));
}
 

@Override
public LongWritable convert( ValueMetaInterface meta, Object obj ) throws TypeConversionException {
  try {
    LongWritable result = new LongWritable();
    result.set( meta.getInteger( obj ) );
    return result;
  } catch ( KettleValueException ex ) {
    throw new TypeConversionException( BaseMessages
      .getString( TypeConverterFactory.class, "ErrorConverting", LongWritable.class.getSimpleName(), obj ), ex );
  }
}
 

public void map(LongWritable key, Text value,
                OutputCollector<LongWritable, Text> output,
                Reporter reporter) throws IOException {
  writeFlag(conf, "map." + name + ".value." + value);
  key.set(10);
  output.collect(key, value);
  if (byValue) {
    assertEquals(10, key.get());
  } else {
    assertNotSame(10, key.get());
  }
  key.set(11);
}
 

public void map(LongWritable key, Text value,
                OutputCollector<LongWritable, Text> output,
                Reporter reporter) throws IOException {
  writeFlag(conf, "map." + name + ".value." + value);
  key.set(10);
  output.collect(key, value);
  if (byValue) {
    assertEquals(10, key.get());
  } else {
    assertNotSame(10, key.get());
  }
  key.set(11);
}
 

/**
 * Internal primitive to write a block-aligned row range of a frame to a single sequence file, 
 * which is used for both single- and multi-threaded writers (for consistency). 
 * 
 * @param path file path
 * @param job job configuration
 * @param fs file system
 * @param src frame block
 * @param blen block length
 * @param rl lower row
 * @param ru upper row
 * @throws IOException if IOException occurs
 */
@SuppressWarnings("deprecation")
protected static void writeBinaryBlockFrameToSequenceFile( Path path, JobConf job, FileSystem fs, FrameBlock src, int blen, int rl, int ru ) 
	throws IOException
{
	//1) create sequence file writer 
	SequenceFile.Writer writer = null;
	writer = new SequenceFile.Writer(fs, job, path, LongWritable.class, FrameBlock.class);
	
	try
	{
		//2) reblock and write
		LongWritable index = new LongWritable();

		if( src.getNumRows() <= blen ) //opt for single block
		{
			//directly write single block
			index.set(1);
			writer.append(index, src);
		}
		else //general case
		{
			//initialize blocks for reuse (at most 4 different blocks required)
			FrameBlock[] blocks = createFrameBlocksForReuse(src.getSchema(), src.getColumnNames(), src.getNumRows());  
			
			//create and write subblocks of frame
			for(int bi = rl; bi < ru; bi += blen) {
				int len = Math.min(blen,  src.getNumRows()-bi);
				
				//get reuse frame block and copy subpart to block (incl meta on first)
				FrameBlock block = getFrameBlockForReuse(blocks);
				src.slice( bi, bi+len-1, 0, src.getNumColumns()-1, block );
				if( bi==0 ) //first block
					block.setColumnMetadata(src.getColumnMetadata());
				
				//append block to sequence file
				index.set(bi+1);
				writer.append(index, block);
			}
		}
	}
	finally {
		IOUtilFunctions.closeSilently(writer);
	}		
}
 

/** write the long value */
static void writeLong(long value, DataOutputStream out) throws IOException {
  LongWritable uLong = TL_DATA.get().U_LONG;
  uLong.set(value);
  uLong.write(out);
}